-
Notifications
You must be signed in to change notification settings - Fork 3
/
ibc.py
160 lines (110 loc) · 3.81 KB
/
ibc.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
"""
cli interface to the trading ibc
"""
import logging
import signal
import os
from argparse import ArgumentParser
from datetime import (
datetime, timedelta
)
import colorama
import ibc.argparser_helpers as argh
import ibc.service as bot
from ibc.console import Console
def run_account(args):
""" Display account status
"""
Console.print('Retrieving account status...')
with bot.Service() as svc:
Console.print('Account: ' + svc.active_account.code)
acc = svc.refresh_account()
Console.print_account(acc)
Console.print_account_positions(acc)
def run_ls(args):
""" Display accounts and orders
"""
Console.print('Listing orders...')
with bot.Service() as svc:
Console.print('Account: ' + svc.active_account.code)
svc.refresh_account()
acc = svc.active_account
Console.print_account(acc)
Console.print_account_positions(acc)
orders = svc.active_account.orders
for order in orders:
Console.print_order(order)
Console.print()
def __run_monitor(svc: bot.Service) -> bool:
""" Run monitor
Returns True on normal exit, False on user quit request
"""
# keyboard controls init
global g_do_exit
g_do_exit = False
def on_press(key):
global g_do_exit
if key in ['q', 'Q']:
g_do_exit = True
Console.print("Terminating...")
# Stop listener
return False
return True
if os.name == 'nt':
from ibc.key_reader import KeyAsyncReader
key_reader = KeyAsyncReader()
key_reader.startReading(on_press)
def signal_handler(signalnum, stackframe):
global g_do_exit
Console.print("Received signal " + str(signalnum) + ", terminating...")
g_do_exit = True
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
if os.name == 'nt':
signal.signal(signal.SIGBREAK, signal_handler)
else:
signal.signal(signal.SIGQUIT, signal_handler)
Console.print_dashboard(svc)
while True:
Console.print('Waiting for updates - Press Q or Ctrl-C to exit...')
time_last_updated = datetime.now()
while True:
svc.broker.wait_for_update(2.)
now = datetime.now()
# prevent very frequent updates
if now - time_last_updated > timedelta(seconds=1):
break
svc.refresh_account()
Console.print_dashboard(svc, delete_previous_lines=True)
if g_do_exit:
break
svc.refresh_account()
Console.print_dashboard(svc, delete_previous_lines=True)
Console.print()
Console.print()
return g_do_exit
def run_monitor(args):
""" Run monitor while there are active trades or continuously
"""
Console.print('Starting monitor...')
with bot.Service() as svc:
Console.print('Account: ' + svc.active_account.code)
__run_monitor(svc)
def main():
argparser = ArgumentParser(add_help=False)
argparser.add_argument('--help', action=argh.EmHelpAction, help='Display help')
argparser.add_argument("-v", "--verbose", dest="verbose", action="store_true", help="Verbose mode")
ap_sp = argparser.add_subparsers(help='sub-command help', required=True)
sp_train = ap_sp.add_parser('account', help='Get account status')
sp_train.set_defaults(func=run_account)
sp_ls = ap_sp.add_parser('ls', help='List trades/orders')
sp_ls.set_defaults(func=run_ls)
sp_monitor = ap_sp.add_parser('monitor', help='Monitor active trades')
sp_monitor.set_defaults(func=run_monitor)
args = argparser.parse_args()
root_logger = logging.getLogger()
root_logger.setLevel(logging.ERROR)
colorama.init()
args.func(args)
if __name__ == "__main__":
main()