MoonBot/test.py at main · agrin96/MoonBot · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
from __future__ import annotations
from typing import List,Any
import numpy as np
import pandas as pd
import json

from Candle import new_candle
from DataBuffer import DataBuffer
from Indicators.Indicators import ChaikinOscillator,MovingAverageConverganceDivergence
from TreeActions import pprint_tree,make_tree_decision,evaluate_next_value
from TreeIO import deserialize_tree

candle_period = 30
prev_volume = 0.0
running_volume = 0

def update_current_candle(candle:Dict,
							new_data:Dict)->bool:
	"""Pass in the current row data to update the current candle. We keep
	updating until we reach the candle period at which point the the candle is
	pushed and a new clean candle is generated. Note we use the best ask price
	in every situation.
	Returns whether the candle has been completed and a new one is needed."""
	global prev_volume
	global running_volume

	live_volume = new_data["total_traded_usdt"] - prev_volume

	prev_volume = new_data["total_traded_usdt"]
	running_volume += live_volume

	if candle["open"] is None:
		candle["open"] = new_data["best_ask"]

	if candle["high"] < new_data["best_ask"]:
		candle["high"] = new_data["best_ask"]

	if candle["low"] > new_data["best_ask"]:
		candle["low"] = new_data["best_ask"]

	candle["volume"] += running_volume
	candle["close"] = new_data["best_ask"]
	candle["elements"] += 1

	if candle["elements"] == candle_period:
		return True

	return False


def main():
	tree = None
	with open("./SerializedTrees/popfile-0.json") as file:
		tree = deserialize_tree(file.read())

	training_data = pd.read_csv("BTCUSDT_ticker.csv")
	training_data = training_data[["best_bid","best_ask","total_traded_usdt"]]
	training_data = training_data.to_dict("records")

	current_candle = new_candle()

	candles = DataBuffer(max_size=500,
						 filename="candles.csv",
						 header=["open","high","low","close","volume","elements"])

	ramp_up_candles = 100
	# ramp_up_candles = 217
	decisions = []
	for data in training_data:
		if update_current_candle(current_candle,data):
			candles.push(current_candle)

			evaluate_next_value(tree,candles.get_all())
			if candles.current_size() > ramp_up_candles:
				decisions.append(make_tree_decision(tree))
			current_candle = new_candle()

	print("Decisions Made: ",len(decisions))
	print(decisions)
	print("Total: ",candles.current_size())

	return decisions


def test_deserialization():
	from TreeIO import deserialize_tree

	with open("./SerializedTrees/popfile-0.json") as file:
		tree = deserialize_tree(file.read())
		print(tree)
		pprint_tree(tree)


if __name__ == "__main__":
	# test_deserialization()
	main()