L6

lab6/currencyData.py

@@ -0,0 +1,54 @@
+import datetime
+
+
+def generate_url(start, end='now', interval="3h", currency='btc-bitcoin'):
+    if end == 'now':
+        end = ''
+    else:
+        end = 'end=%s&' % end
+    return 'https://api.coinpaprika.com/v1/tickers/%s/historical?start=%s&%slimit=5000&interval=%s' \
+           % (currency, start, end, interval)
+
+
+def prepare_values(data):
+    dates, prices, volumes = [], [], []
+    for node in data:
+        formatted_date = datetime.datetime \
+            .strptime(node.get('timestamp'), '%Y-%m-%dT%H:%M:%SZ')
+        dates.append(formatted_date)
+        prices.append(node.get('price'))
+        volumes.append(node.get('volume_24h') / 10000000)
+    return dates, prices, volumes
+
+
+def generate_dates(first, time_step, elements):
+    base = datetime.datetime.strptime(first, '%Y-%m-%d %H:%M:%S')
+    dates = []
+    for x in range(elements):
+        dates.append(base)
+        base = base + time_step
+    return dates
+
+
+def count_average(values, fun):
+    to_return = []
+    for value in values:
+        to_return.append(fun(value))
+    return to_return
+
+
+def generate_future_values(forecaster):
+    all_prices, all_volumes = [], []
+    for i in range(forecaster.total):
+        all_prices.append([])
+        all_volumes.append([])
+    for i in range(100):
+        new_prices, new_volumes = forecaster.forecast()
+        for j in range(len(new_prices)):
+            all_prices[j].append(new_prices[j])
+            all_volumes[j].append(new_volumes[j])
+    return all_prices, all_volumes
+
+
+def generate_stats(all_prices, all_volumes, fun):
+    return count_average(all_prices, fun), count_average(all_volumes, fun)

lab6/forecaster.py

@@ -0,0 +1,114 @@
+import random
+import statistics
+from datetime import datetime
+random.seed(datetime.now())
+
+
+increasing = True
+decreasing = False
+
+
+class VolumeData:
+    def __init__(self, price_up, price_down):
+        self.price_up = price_up
+        self.price_down = price_down
+
+
+def if_modify(prob):
+    return random.random() < prob
+
+
+class Generator:
+    def __init__(self, prices, volumes):
+        assert len(prices) == len(volumes)
+        self.prices = prices
+        self.volumes = volumes
+        self.price_diffs = []
+        self.volume_diffs = []
+        self.total = len(prices)
+        self.volumes_down = []
+        self.volumes_up = []
+        self.volume_trend_keeps_up = []
+        self.volume_trend_keeps_down = []
+        self.volume_trend_change_to_up = []
+        self.volume_trend_change_to_down = []
+        self.volume_average = statistics.mean(self.volumes)
+        self.volumes_dev = statistics.stdev(self.volumes)
+        self.generate_stats()
+
+    def generate_stats(self):
+        volume_trend = increasing  # need to initialize somehow
+        self.price_diffs.append(0.0)
+        self.volume_diffs.append(0.0)
+        for i in range(1, len(self.prices)):
+            self.price_diffs.append(self.prices[i] - self.prices[i - 1])
+            self.volume_diffs.append(self.volumes[i] - self.volumes[i - 1])
+            if self.volume_diffs[i] > 0:
+                self.volumes_up.append(self.price_diffs[i] / self.prices[i])
+                if volume_trend == increasing:
+                    self.volume_trend_keeps_up.append(self.volume_diffs[i] / self.volumes[i])
+                else:
+                    self.volume_trend_change_to_up.append(self.volume_diffs[i] / self.volumes[i])
+                    volume_trend = increasing
+            else:
+                self.volumes_down.append(self.price_diffs[i] / self.prices[i])
+                if volume_trend == decreasing:
+                    self.volume_trend_keeps_down.append(self.volume_diffs[i] / self.volumes[i])
+                else:
+                    self.volume_trend_change_to_down.append(self.volume_diffs[i] / self.volumes[i])
+                    volume_trend = decreasing
+
+    def generate_price(self, volume_trend, current_price):
+        if volume_trend == increasing:
+            cases = self.volumes_up
+        else:
+            cases = self.volumes_down
+        return cases[random.randrange(len(cases))] * current_price + current_price
+
+    def regulate_decrease(self, volumes, index):
+        if index == 0:
+            return 0.0
+        standard_min = self.volume_average - self.volumes_dev
+        prob = 0.0
+        if standard_min > volumes:
+            prob = 1 - (volumes / standard_min)
+        return prob
+
+    def generate_volume(self, current_volume, current_trend):
+        if current_trend == increasing:
+            volume_increase = self.volume_trend_keeps_up
+            volume_decrease = self.volume_trend_change_to_down
+        else:
+            volume_decrease = self.volume_trend_keeps_down
+            volume_increase = self.volume_trend_change_to_up
+        total_cases = len(volume_increase) + len(volume_decrease)
+        bound = len(volume_increase)
+        index = random.randrange(total_cases)
+        modify_prob = self.regulate_decrease(current_volume, index)
+        if if_modify(modify_prob):
+            index = random.randrange(index)
+        if index < bound:
+            diff = volume_increase[index]
+        else:
+            index = index - bound
+            diff = volume_decrease[index]
+        new_volume = current_volume * diff + current_volume
+        return new_volume
+
+    def forecast(self):
+        current_volume = self.volumes[-1]
+        current_price = self.prices[-1]
+        volume_trend = self.volume_diffs[-1] > 0
+        future_volumes = []
+        future_prices = []
+        for i in range(0, self.total):
+            new_volume = self.generate_volume(current_volume, volume_trend)
+            volume_trend = new_volume > current_volume
+            future_volumes.append(current_volume)
+            current_volume = new_volume
+            new_price = self.generate_price(volume_trend, current_price)
+            future_prices.append(new_price)
+            current_price = new_price
+        return future_prices, future_volumes
+
+

lab6/main.py

@@ -0,0 +1,44 @@
+import requests
+import datetime
+from statistics import mean, stdev, median
+import matplotlib.pyplot as plt
+from currencyData import generate_url, prepare_values, generate_dates, generate_future_values, generate_stats
+from forecaster import Generator
+
+
+def print_price_chart(first_series, second_series):
+    x, y, bars = first_series
+    plt.xlabel('date')
+    plt.ylabel('price[USD], volume[10^7]')
+    plt.xticks(rotation=60)
+    plt.plot(x, y)
+    plt.bar(x, bars, color=['green', 'blue'])
+    x, y, bars = second_series
+    plt.plot(x, y)
+    plt.bar(x, bars, color=['red', 'orange'])
+    plt.show()
+
+
+def print_bar_char(data, time, title):
+    plt.xticks(rotation=60)
+    plt.title(title)
+    plt.bar(time, data)
+    plt.show()
+
+
+if __name__ == '__main__':
+    url = generate_url('2018-03-29', '2018-04-29', interval='3h')
+    data = requests.get(url).json()
+    dates, prices, volumes = prepare_values(data)
+    model = Generator(prices, volumes)
+    future_prices, future_volumes = generate_future_values(model)
+    future_dates = generate_dates('2018-04-29 00:00:00', datetime.timedelta(hours=3), len(dates))
+    average_prices, average_volumes = generate_stats(future_prices, future_volumes, mean)
+    print_price_chart((dates, prices, volumes), (future_dates, average_prices, average_volumes))
+    price_dev, volume_dev = generate_stats(future_prices, future_volumes, stdev)
+    price_median, volume_median = generate_stats(future_prices, future_volumes, median)
+    print_bar_char(price_dev, future_dates, 'price deviations')
+    print_bar_char(volume_dev, future_dates, 'volumes deviations')
+    print_bar_char(price_median, future_dates, 'price medians')
+    print_bar_char(volume_median, future_dates, 'volume medians')
+