forked from robcarver17/pysystemtrade
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmongo_position_limits.py
135 lines (102 loc) · 4.51 KB
/
mongo_position_limits.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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
from sysdata.mongodb.mongo_generic import mongoDataWithMultipleKeys
from sysdata.production.position_limits import positionLimitData
from sysobjects.production.position_limits import (
positionLimitForInstrument,
positionLimitForStrategyInstrument,
)
from sysobjects.production.tradeable_object import (
listOfInstrumentStrategies,
instrumentStrategy,
)
from syslogging.logger import *
POSITION_LIMIT_STATUS_COLLECTION = "position_limit_status"
MARKER_KEY = "marker"
MARKER_STRATEGY_INSTRUMENT = "strategy_instrument"
MARKER_INSTRUMENT = "instrument"
INSTRUMENT_KEY = "instrument_code"
STRATEGY_KEY = "strategy_name"
POSITION_LIMIT_KEY = "position_limit"
class mongoPositionLimitData(positionLimitData):
"""
Read and write data class to get override state data
"""
def __init__(self, mongo_db=None, log=get_logger("mongoPositionLimitData")):
super().__init__(log=log)
self._mongo_data = mongoDataWithMultipleKeys(
POSITION_LIMIT_STATUS_COLLECTION, mongo_db=mongo_db
)
@property
def mongo_data(self):
return self._mongo_data
def __repr__(self):
return "Data connection for position limit data, mongodb %s"
def get_all_instruments_with_limits(self) -> list:
dict_of_keys = {MARKER_KEY: MARKER_INSTRUMENT}
list_of_dicts = self.mongo_data.get_list_of_result_dicts_for_dict_keys(
dict_of_keys
)
list_of_instruments = [db_entry[INSTRUMENT_KEY] for db_entry in list_of_dicts]
return list_of_instruments
def get_all_instrument_strategies_with_limits(self) -> listOfInstrumentStrategies:
dict_of_keys = {MARKER_KEY: MARKER_STRATEGY_INSTRUMENT}
list_of_dicts = self.mongo_data.get_list_of_result_dicts_for_dict_keys(
dict_of_keys
)
list_of_instrument_strategies = [
instrumentStrategy(
strategy_name=db_entry[STRATEGY_KEY],
instrument_code=db_entry[INSTRUMENT_KEY],
)
for db_entry in list_of_dicts
]
list_of_instrument_strategies = listOfInstrumentStrategies(
list_of_instrument_strategies
)
return list_of_instrument_strategies
def delete_position_limit_for_instrument_strategy(
self, instrument_strategy: instrumentStrategy
):
dict_of_keys = {
MARKER_KEY: MARKER_STRATEGY_INSTRUMENT,
STRATEGY_KEY: instrument_strategy.strategy_name,
INSTRUMENT_KEY: instrument_strategy.instrument_code,
}
self.mongo_data.delete_data_without_any_warning(dict_of_keys)
def delete_position_limit_for_instrument(self, instrument_code: str):
dict_of_keys = {MARKER_KEY: MARKER_INSTRUMENT, INSTRUMENT_KEY: instrument_code}
self.mongo_data.delete_data_without_any_warning(dict_of_keys)
def _get_abs_position_limit_for_instrument_strategy(
self, instrument_strategy: instrumentStrategy
) -> int:
dict_of_keys = {
MARKER_KEY: MARKER_STRATEGY_INSTRUMENT,
STRATEGY_KEY: instrument_strategy.strategy_name,
INSTRUMENT_KEY: instrument_strategy.instrument_code,
}
find_object_dict = self.mongo_data.get_result_dict_for_dict_keys(dict_of_keys)
position_limit = find_object_dict[POSITION_LIMIT_KEY]
return position_limit
def _get_abs_position_limit_for_instrument(
self,
instrument_code: str,
) -> int:
dict_of_keys = {MARKER_KEY: MARKER_INSTRUMENT, INSTRUMENT_KEY: instrument_code}
find_object_dict = self.mongo_data.get_result_dict_for_dict_keys(dict_of_keys)
position_limit = find_object_dict[POSITION_LIMIT_KEY]
return position_limit
def set_position_limit_for_instrument_strategy(
self, instrument_strategy: instrumentStrategy, new_position_limit: int
):
dict_of_keys = {
MARKER_KEY: MARKER_STRATEGY_INSTRUMENT,
STRATEGY_KEY: instrument_strategy.strategy_name,
INSTRUMENT_KEY: instrument_strategy.instrument_code,
}
data_dict = {POSITION_LIMIT_KEY: new_position_limit}
self.mongo_data.add_data(dict_of_keys, data_dict, allow_overwrite=True)
def set_position_limit_for_instrument(
self, instrument_code: str, new_position_limit: int
):
dict_of_keys = {MARKER_KEY: MARKER_INSTRUMENT, INSTRUMENT_KEY: instrument_code}
data_dict = {POSITION_LIMIT_KEY: new_position_limit}
self.mongo_data.add_data(dict_of_keys, data_dict, allow_overwrite=True)