This repository has been archived by the owner on Apr 12, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 120
/
event_types.py
166 lines (133 loc) · 5.41 KB
/
event_types.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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
"""
trading-server is a multi-asset, multi-strategy, event-driven execution
and backtesting platform (OEMS) for trading common markets.
Copyright (C) 2020 Sam Breznikar <sam@sdbgroup.io>
Licensed under GNU General Public License 3.0 or later.
Some rights reserved. See LICENSE.md, AUTHORS.md.
"""
from dateutil import parser
from datetime import datetime
class Event(object):
"""
Base class for system events.
"""
class MarketEvent(Event):
"""
Wrapper for new market data. Consumed by Strategy object to
produce Signal events.
"""
# Datetime object format string
DTFMT = '%Y-%m-%d %H:%M'
def __init__(self, exchange, bar):
self.type = 'MARKET'
self.exchange = exchange
self.bar = bar
def __str__(self):
return str("MarketEvent - Exchange: " + self.exchange.get_name() +
" Symbol: " + self.bar['symbol'] + " TS: " +
self.get_datetime() + " Close: " + self.bar['close'])
def get_bar(self):
return self.bar
def get_exchange(self):
return self.exchange
def get_datetime(self):
return datetime.fromtimestamp(
self.bar['timestamp']).strftime(self.DTFMT),
class SignalEvent(Event):
"""
Entry signal. Consumed by Portfolio to produce Order events.
"""
def __init__(self, symbol: str, entry_ts, direction: str, timeframe: str,
strategy: str, venue, entry_price: float, entry_type: str,
targets: list, stop_price: float, void_price: float,
trail: bool, note: str, dataset, ic=1):
self.type = 'SIGNAL'
self.entry_ts = entry_ts # Entry bar timestamp.
self.timeframe = timeframe # Signal timeframe.
self.strategy = strategy # Signal strategy name.
self.venue = venue # Signal venue name.
self.symbol = symbol # Ticker code for instrument.
self.direction = direction # LONG or SHORT.
self.entry_price = entry_price # Trade entry price.
self.entry_type = entry_type.upper() # Order type for entry.
self.targets = targets # [(price target, int % to close)]
self.stop_price = stop_price # Stop-loss order price.
self.void_price = void_price # Invalidation price.
self.instrument_count = ic # # of instruments in use.
self.trail = trail # True or False for trailing stop.
self.op_data = dataset # Dataset used to generate signal.
self.note = note # Signal notes.
def __str__(self):
return str("Signal Event: " + self.direction + " Symbol: " +
self.symbol + " Entry price: " + str(self.entry_price) +
" Entry timestamp: " + str(self.entry_ts) + " Timeframe: " +
self.timeframe + " Strategy: " + self.strategy +
" Venue: " + self.venue.get_name() + " Order type: " +
self.entry_type + " Note: " + self.note)
def get_signal_dict(self):
return {
'strategy': self.strategy,
'venue': self.venue.get_name(),
'symbol': self.symbol,
'entry_timestamp': self.entry_ts,
'timeframe': self.timeframe,
'direction': self.direction,
'entry_price': self.entry_price,
'entry_type': self.entry_type,
'targets': self.targets,
'stop_price': self.stop_price,
'void_price': self.void_price,
'instrument_count': self.instrument_count,
'trail': self.trail,
'note': self.note,
'op_data': self.op_data}
def inverse_direction(self):
"""
Return the opposite direction of 'direction' variable.
"""
if self.direction.upper() == "LONG":
return "SHORT"
elif self.direction.upper() == "SHORT":
return "LONG"
class OrderEvent(Event):
"""
Contains trade details to be sent to a broker/exchange.
"""
def __init__(self, order_dict):
self.type = 'ORDER'
self.order_dict = order_dict
self.trade_id = order_dict['trade_id']
self.order_id = order_dict['order_id']
self.timestamp = order_dict['timestamp']
self.avg_fill_price = order_dict['avg_fill_price']
self.currency = order_dict['currency']
self.venue_id = order_dict['venue_id']
self.direction = order_dict['direction']
self.size = order_dict['size']
self.price = order_dict['price']
self.order_type = order_dict['order_type']
self.metatype = order_dict['metatype']
self.void_price = order_dict['void_price']
self.trail = order_dict['trail']
self.reduce_only = order_dict['reduce_only']
self.post_only = order_dict['post_only']
self.batch_size = order_dict['batch_size']
self.status = order_dict['status']
def __str__(self):
return str(" ")
def get_order_dict(self):
return self.order_dict
class FillEvent(Event):
"""
Holds transaction data including fees/commissions, slippage, brokerage,
actual fill price, timestamp, etc.
"""
def __init__(self, order_conf):
self.type = 'FILL'
self.order_conf = order_conf
# TODO
self.fees = None
def __str__(self):
return str(" ")
def get_order_conf(self):
return self.order_conf