forked from moonnejs/PythonLab
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathctaBasicModel.py
More file actions
155 lines (132 loc) · 5.63 KB
/
ctaBasicModel.py
File metadata and controls
155 lines (132 loc) · 5.63 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
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
# encoding: UTF-8
'''
本文件中包含了CTA模块中用到的一些基础设置、类和常量等。
'''
from __future__ import division
from qtpy import QtGui, QtCore
from ctaBase import *
from eventEngine import *
########################################################################
class StrategyBacktesting(QtGui.QStandardItemModel):
"""用于回测的策略信息"""
signal = QtCore.Signal(type(Event()))
#----------------------------------------------------------------------
def __init__(self, eventEngine, ctaEngine=None, view = None):
"""Constructor"""
super(StrategyBacktesting, self).__init__(None)
self.eventEngine = eventEngine
self.ctaEngine = ctaEngine
self.view = view
self.view = view
self.nRow = 0
self.nameItems = {}
self.classItems = {}
self.initUi()
self.updateData()
#----------------------------------------------------------------------
def initUi(self):
""""""
self.setHorizontalHeaderItem(0, QtGui.QStandardItem(u'策略信息'))
self.setHorizontalHeaderItem(1, QtGui.QStandardItem(u'交易合约'))
self.setHorizontalHeaderItem(2, QtGui.QStandardItem(u'使用资金'))
#----------------------------------------------------------------------
def registerEvent(self):
"""注册事件监听"""
self.signal.connect(self.updateData)
self.eventEngine.register(EVENT_CTA_STRATEGY, self.signal.emit)
#----------------------------------------------------------------------
def updateData(self,name0=''):
if self.nRow > 0:
self.removeRows(0,self.nRow)
self.classItems = {}
self.nameItems = {}
for name in self.ctaEngine.strategyDict:
paramDict = self.ctaEngine.getStrategyParam(name)
className = paramDict['className']
if not className in self.classItems:
self.classItems[className] = QtGui.QStandardItem(paramDict['className'])
self.appendRow([self.classItems[className]])
nameItem = QtGui.QStandardItem(name)
self.nameItems[name] = nameItem
if name != name0:
nameItem.setCheckState(False)
else:
nameItem.setCheckState(True)
nameItem.setCheckable(False)
self.classItems[className].appendRow([
nameItem,
QtGui.QStandardItem(str(paramDict['symbolList'])),
QtGui.QStandardItem(str(paramDict['capital']))])
self.nRow = len(self.classItems.values())
if self.view: self.view.expandAll()
#----------------------------------------------------------------------
def checkName(self,name0):
for name in self.nameItems:
if name0==name:
self.nameItems[name].setCheckState(True)
else:
self.nameItems[name].setCheckState(False)
########################################################################
class StrategyParam(QtGui.QStandardItemModel):
"""用于回测的策略信息"""
signal = QtCore.Signal(type(Event()))
#----------------------------------------------------------------------
def __init__(self, eventEngine, ctaEngine=None):
"""Constructor"""
super(StrategyParam, self).__init__(None)
self.eventEngine = eventEngine
self.ctaEngine = ctaEngine
self.name = ''
self.nRow = 0
self.nameItems = {}
#----------------------------------------------------------------------
def registerEvent(self):
"""注册事件监听"""
self.signal.connect(self.updateData)
self.eventEngine.register(EVENT_CTA_STRATEGY, self.signal.emit)
#----------------------------------------------------------------------
def updateData(self,name,append = False):
if not append:
#self.removeRows(0,self.nRow)
self.clear()
self.nRow = 1
else:
self.nRow += 1
paramDict = self.ctaEngine.getStrategyParam(name)
keys = []
values = []
for k,v in paramDict.items():
keys.append(k)
values.append(v)
for i in range(len(keys)):
self.setHorizontalHeaderItem(i, QtGui.QStandardItem(str(keys[i])))
self.appendRow([
QtGui.QStandardItem(str(v)) for v in values
])
########################################################################
class MongoData(QtGui.QStandardItemModel):
"""数据信息"""
signal = QtCore.Signal(type(Event()))
#----------------------------------------------------------------------
def __init__(self, eventEngine, ctaEngine=None):
"""Constructor"""
super(MongoData, self).__init__(None)
self.eventEngine = eventEngine
self.ctaEngine = ctaEngine
self.name = ''
self.nRow = 0
self.nameItems = {}
#----------------------------------------------------------------------
def registerEvent(self):
"""注册事件监听"""
self.signal.connect(self.updateData)
self.eventEngine.register(EVENT_CTA_STRATEGY, self.signal.emit)
#----------------------------------------------------------------------
def updateData(self,name):
if self.nRow > 0:
self.removeRows(0,self.nRow)
self.setHorizontalHeaderItem(0, QtGui.QStandardItem(u'合约'))
self.setHorizontalHeaderItem(1, QtGui.QStandardItem(u'数据范围'))
self.setHorizontalHeaderItem(2, QtGui.QStandardItem(u'文件大小'))
self.appendRow([
QtGui.QStandardItem(str(v)) for v in values])