# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# +
import ssl

import tensorflow as tf
import pandas as pd
import numpy as np

ssl._create_default_https_context = ssl._create_unverified_context

csv_file = tf.keras.utils.get_file('Coinbase_BTCUSD_1h.csv', 'https://www.cryptodatadownload.com/cdd/Coinbase_BTCUSD_1h.csv')
data_frame = pd.read_csv(csv_file, skiprows=1, index_col="Date")

data_frame.index = pd.to_datetime(data_frame.index, format='%Y-%m-%d %I-%p')

data_frame.head()

# +
prepared_df = data_frame.drop('Symbol', axis=1)
prepared_df['Date'] = prepared_df.index.values.astype(np.int64) // 10 ** 9
prepared_df = prepared_df.sort_index()

print(prepared_df.head())

train_dataset = tf.data.Dataset.from_tensor_slices(dict(prepared_df))

print([x['Close'].numpy() for x in train_dataset.take(5)])

# +
ds_iter = iter(train_dataset)

while True:
    batch = next(ds_iter)
    print(batch['Close'])

# +
from sklearn.model_selection import train_test_split

train_df, test_df = train_test_split(prepared_df, test_size=0.2)

volume = tf.feature_column.numeric_column("Volume USD")
volume_buckets = tf.feature_column.bucketized_column(volume, [1000, 10000, 50000, 100000, 250000, 500000, 1000000, 2500000, 5000000, 10000000])

feature_layer = tf.keras.layers.DenseFeatures(volume_buckets)

train_labels = feature_layer(dict(train_df)).numpy()
test_labels = feature_layer(dict(test_df)).numpy()

train_labels, test_labels

# +
no_volume_train = train_df.drop(['Volume BTC', 'Volume USD'], axis=1)
no_volume_test = test_df.drop(['Volume BTC', 'Volume USD'], axis=1)

train_ds = tf.data.Dataset.from_tensor_slices((no_volume_train.values, train_labels)).shuffle(1024).batch(32)
test_ds = tf.data.Dataset.from_tensor_slices((no_volume_test.values, test_labels)).shuffle(1024).batch(32)

model = tf.keras.Sequential([
  tf.keras.layers.Flatten(),
  tf.keras.layers.Dense(128),
  tf.keras.layers.Dense(64),
  tf.keras.layers.Dense(11),
])

model.compile(optimizer='adam',
              loss=tf.keras.losses.CategoricalCrossentropy(), 
              metrics=['accuracy'])

model.fit(train_ds)

loss, accuracy = model.evaluate(test_ds)

# +
import tensorflow as tf
import numpy as np

from IPython.display import SVG, Image
from tensorflow.keras.utils import plot_model

items_per_gen = 1000
n_samples = 100
output_shape=(items_per_gen, 1)

generator = tf.keras.Sequential([
            tf.keras.layers.Dense(units=256, input_shape=(1, n_samples), activation="relu"),
            tf.keras.layers.LeakyReLU(alpha=0.2),
            tf.keras.layers.BatchNormalization(momentum=0.8),
            tf.keras.layers.Dense(512),
            tf.keras.layers.LeakyReLU(alpha=0.2),
            tf.keras.layers.BatchNormalization(momentum=0.8),
            tf.keras.layers.Dense(1024),
            tf.keras.layers.LeakyReLU(alpha=0.2),
            tf.keras.layers.BatchNormalization(momentum=0.8),
            tf.keras.layers.Dense(np.prod(output_shape), activation='tanh'),
            tf.keras.layers.Reshape(output_shape)
        ])

discriminator = tf.keras.Sequential([
    tf.keras.layers.LSTM(512, input_shape=output_shape, return_sequences=True),
    tf.keras.layers.Bidirectional(tf.keras.layers.LSTM(512)),
    tf.keras.layers.Dense(512),
    tf.keras.layers.LeakyReLU(alpha=0.2),
    tf.keras.layers.Dense(256),
    tf.keras.layers.LeakyReLU(alpha=0.2),
    tf.keras.layers.Dense(1, activation='sigmoid'),
])
# -

plot_model(generator, to_file='test_keras_plot_model.png', show_shapes=True)
Image('test_keras_plot_model.png')

plot_model(discriminator, to_file='test_keras_plot_model_1.png', show_shapes=True)
Image('test_keras_plot_model_1.png')

StarcoderdataJupyterScriptsDedupFiltered