tradingshell/shell.py at master · xttjsn/tradingshell · 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
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
167
168
169
170
171
172
173
174
175
176
"""The shell itself
"""
import os
import json
import argparse
import pdb
from base import (MyCmd,
                  with_argparser,
                  pipeReader,
                  MSG_SEP)
from service import WebDisplayService
from backtest import ZiplineRunThread
import time

DEFAULT_PLOT_NAME = '<default_plot>'


class TradingShell(MyCmd):

    prompt = '>>>'

    def __init__(self):
        super(TradingShell, self).__init__()
        self.backtest_engines = ['zipline']
        self.actions = {
            'backtest': self.backtest,
            'echo': self.echo,
            'plot': self.plot,
            'service': self.service,
            'db': self.db
        }
        self.services = {}

    def get_service(self, name):
        try:
            service = self.services[name]
        except Exception:
            service = WebDisplayService()
            self.services[name] = service
        return service

    def do_b(self, arg):
        self.do_backtest(arg)

    def do_backtest(self, arg):
        self.piperun('backtest ' + arg)

    argparser = argparse.ArgumentParser()
    argparser.add_argument('engine', default='zipline')
    argparser.add_argument('algofile', default='./algo.py')
    @with_argparser(argparser)
    def backtest(self, args, inPipe=None, outPipe=1):
        if args.engine not in self.backtest_engines:
            self.poutput('{} not a valid backtest engine'.format(args.engine))
            return

        with open(args.algofile, 'r') as f:
            code = f.read()

        r, w = os.pipe()

        def consume_portfolio(portfolio):
            update = json.dumps({
                'portfolio_value': portfolio['portfolio_value'],
                'pnl': portfolio['pnl'],
                'return': portfolio['returns']
            }).encode('utf8') + MSG_SEP
            os.write(outPipe, update)

        ziplineTh = ZiplineRunThread(code, '2012-01-01', '2012-6-01', 100000,
                                     consume_portfolio)
        ziplineTh.start()

    def do_p(self, arg):
        self.do_plot(arg)

    def do_plot(self, arg):
        pdb.set_trace()
        self.piperun('plot ' + arg)

    argparser = argparse.ArgumentParser()
    argparser.add_argument('frontend', default='web')
    @with_argparser(argparser)
    def plot(self, arg, inPipe=None, outPipe=1):
        """Plot the current data according to its type
        """

        if arg.frontend == 'web':
            service = self.get_service('webdisplay')

            if inPipe is not None:
                initParams = {
                    'title': 'Default Plot',
                    'xlabel': 'Date',
                    'startDate': '2012-01-01'
                }

                service.start()
                self.poutput('Waiting for service to start...')
                time.sleep(3)

                initParams = {
                    'title': 'Default Plot',
                    'xlabel': 'Date',
                    'startDate': '2012-01-01'
                }
                service.updateAddPlotter(pipeReader(inPipe, MSG_SEP),
                                         initParams)

            else:
                raise Exception('Non-pipe data model not implemented!')

        elif arg.frontend == 'plt':
            raise Exception('plt front end not implemented yet!')

    def do_s(self, arg):
        self.do_service(arg)

    def do_service(self, arg):
        self.piperun('service ' + arg)

    argparser = argparse.ArgumentParser()
    argparser.add_argument('service', default='webdisplay')
    argparser.add_argument('action', default='status')
    @with_argparser(argparser)
    def service(self, arg, inPipe=None, outPipe=1):
        """Manage services
        """
        # Run backtest as a service, once per day.
        # Make it manageable through this shell.
        # I.e. store the PID in database somewhere
        if arg.service == 'webdisplay':
            service = self.get_service('webdisplay')

            if arg.action == 'start':
                service.start()

            elif arg.action == 'status':
                self.poutput(service.status())

            elif arg.action == 'stop':
                service.stop()

            elif arg.action == 'update':
                if inPipe is None:
                    return

                initParams = {
                    'title': 'Default Plot',
                    'xlabel': 'Date',
                    'startDate': '2012-01-01'
                }
                service.updateAddPlotter(pipeReader(inPipe, MSG_SEP),
                                         initParams)

            else:
                self.perror('Invalid action {}.'.format(arg.action))

        else:
            self.perror('Not yet implemented: {}.'.format(arg.service))

    def do_db(self, arg):
        self.piperun('db ' + arg)

    def do_db_hello(self, arg):
        pass

    argparser = argparse.ArgumentParser()

    def db(self, arg, inPipe=None, outPipe=1):
        pass


if __name__ == '__main__':
    shell = TradingShell()
    shell.cmdloop()