forked from robcarver17/pysystemtrade
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathcsv_historic_orders.py
83 lines (63 loc) · 2.6 KB
/
csv_historic_orders.py
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
import pandas as pd
from sysdata.production.historic_orders import (
strategyHistoricOrdersData,
contractHistoricOrdersData,
)
from syscore.fileutils import resolve_path_and_filename_for_package
from syscore.constants import arg_not_supplied
from syslogging.logger import *
DATE_INDEX_NAME = "DATETIME"
def from_list_of_orders_to_df(list_of_orders):
list_as_dict = [order.as_dict() for order in list_of_orders]
if len(list_of_orders) > 0:
keys = list(list_as_dict[-1].keys())
else:
keys = []
key_data = {}
for key_name in keys:
key_data[key_name] = [dict_entry[key_name] for dict_entry in list_as_dict]
index_data = [dict_entry["fill_datetime"] for dict_entry in list_as_dict]
orders_as_df = pd.DataFrame(key_data, index=index_data)
return orders_as_df
class csvStrategyHistoricOrdersData(strategyHistoricOrdersData):
def __init__(
self, datapath=arg_not_supplied, log=get_logger("csvStrategyPositionData")
):
super().__init__(log=log)
if datapath is None:
raise Exception("Need to provide datapath")
self._datapath = datapath
def write_orders(self, list_of_orders):
filename = resolve_path_and_filename_for_package(
self._datapath, "%s.csv" % ("strategy_orders")
)
df = from_list_of_orders_to_df(list_of_orders)
df.to_csv(filename, index_label=DATE_INDEX_NAME)
class csvContractHistoricOrdersData(contractHistoricOrdersData):
def __init__(
self, datapath=arg_not_supplied, log=get_logger("csvContractPositionData")
):
super().__init__(log=log)
if datapath is None:
raise Exception("Need to provide datapath")
self._datapath = datapath
def write_orders(self, list_of_orders):
filename = resolve_path_and_filename_for_package(
self._datapath, "%s.csv" % ("contract_orders")
)
df = from_list_of_orders_to_df(list_of_orders)
df.to_csv(filename, index_label=DATE_INDEX_NAME)
class csvBrokerHistoricOrdersData(contractHistoricOrdersData):
def __init__(
self, datapath=arg_not_supplied, log=get_logger("csvBrokerHistoricOrdersData")
):
super().__init__(log=log)
if datapath is None:
raise Exception("Need to provide datapath")
self._datapath = datapath
def write_orders(self, list_of_orders):
filename = resolve_path_and_filename_for_package(
self._datapath, "%s.csv" % ("broker_orders")
)
df = from_list_of_orders_to_df(list_of_orders)
df.to_csv(filename, index_label=DATE_INDEX_NAME)