forked from QuantFans/quantdigger
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathwindowgate.py
More file actions
89 lines (71 loc) · 2.6 KB
/
Copy pathwindowgate.py
File metadata and controls
89 lines (71 loc) · 2.6 KB
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
# -*- coding: utf-8 -*-
##
# @file windowgate.py
# @brief
# @author wondereamer
# @version 0.5
# @date 2016-07-10
from quantdigger.datastruct import PContract
from quantdigger.event.rpc import EventRPCClient, EventRPCServer
from quantdigger.event.eventengine import ZMQEventEngine
from quantdigger.interaction.interface import BackendInterface
from quantdigger.util import gen_logger as log
from quantdigger.interaction.backend import Backend
from quantdigger.interaction.serialize import (
deserialize_all_pcontracts,
deserialize_all_contracts,
deserialize_pcontract_bars
)
class WindowGate(BackendInterface):
SERVER_FOR_SHELL = "ui4shell"
def __init__(self, widget):
log.info("Init WindowGate..")
self._engine = ZMQEventEngine('WindowGate')
self._engine.start()
self._backend = EventRPCClient('WindowGate', self._engine, Backend.SERVER_FOR_UI)
self._shell_srv = EventRPCServer(self._engine, self.SERVER_FOR_SHELL)
self._register_functions(self._shell_srv)
self._period = None
self._contract = None
self._widget = widget
def _register_functions(self, server):
server.register('get_all_contracts', self.get_all_contracts)
server.register('get_all_pcontracts', self.get_all_pcontracts)
server.register('get_pcontract', self.get_pcontract)
server.register('show_data', self.show_data)
def add_widget(self, ith, type_):
self._widget.add_widget
def add_technical(self, ith, technical):
""""""
# @TODO compute technical with backend,
# display result from backend
return
@property
def pcontract(self):
return PContract(self._contract, self._period)
def stop(self):
self._engine.stop()
def get_all_contracts(self):
ret = self._backend.sync_call('get_all_contracts')
return deserialize_all_contracts(ret)
def get_all_pcontracts(self):
ret = self._backend.sync_call('get_all_pcontracts')
return deserialize_all_pcontracts(ret)
def get_pcontract(self, str_pcontract):
ret = self._backend.sync_call('get_pcontract', {
'str_pcontract': str_pcontract
})
return deserialize_pcontract_bars(ret)
def run_strategy(self, name):
""""""
return
def run_technical(self, name):
return
def get_technicals(self):
""" 获取系统的所有指标。 """
return
def get_strategies(self):
""" 获取系统的所有策略。 """
return
def show_data(self, pcontract):
self._widget.show_data(pcontract)