senset/vnpy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[新功能] 策略引擎增强版

Browse Source
This commit is contained in:
msincenselee 2020-01-01 22:44:35 +08:00
parent e9c1f80890
commit c499bd929a
8 changed files with 2192 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
vnpy/app/cta_strategy_pro/__init__.py Normal file
View File

@ -0,0 +1,22 @@
from pathlib import Path
from vnpy.trader.app import BaseApp
from vnpy.trader.constant import Direction
from vnpy.trader.object import TickData, BarData, TradeData, OrderData
from vnpy.trader.utility import BarGenerator, ArrayManager
from .base import APP_NAME, StopOrder
from .engine import CtaEngine
from .template import CtaTemplate, CtaSignal, TargetPosTemplate
class CtaStrategyProApp(BaseApp):
""""""
app_name = APP_NAME
app_module = __module__
app_path = Path(__file__).parent
display_name = "CTA策略PRO"
engine_class = CtaEngine
widget_name = "CtaManager"
icon_name = "cta.ico"

93
vnpy/app/cta_strategy_pro/base.py Normal file
View File

@ -0,0 +1,93 @@
"""
Defines constants and objects used in CtaStrategyPro App.
"""
from dataclasses import dataclass, field
from enum import Enum
from datetime import timedelta
from vnpy.trader.constant import Direction, Offset, Interval
APP_NAME = "CtaStrategyPro"
STOPORDER_PREFIX = "STOP"
class StopOrderStatus(Enum):
WAITING = "等待中"
CANCELLED = "已撤销"
TRIGGERED = "已触发"
class EngineType(Enum):
LIVE = "实盘"
BACKTESTING = "回测"
class BacktestingMode(Enum):
BAR = 1
TICK = 2
class Area(Enum):
""" Kline area """
LONG_A = 'LONG_A'
LONG_B = 'LONG_B'
LONG_C = 'LONG_C'
LONG_D = 'LONG_D'
LONG_E = 'LONG_E'
SHORT_A = 'SHORT_A'
SHORT_B = 'SHORT_B'
SHORT_C = 'SHORT_C'
SHORT_D = 'SHORT_D'
SHORT_E = 'SHORT_E'
# 各类商品所在市场，underly_symbol: price_tick
# 上期所夜盘，9:00~10:15, 10:30~11:30, 13:30~15:00, 21:00 ~2:30
NIGHT_MARKET_SQ1 = {'AU': 0.05, 'AG': 1, 'SC': 0.1}
# 上期所夜盘，9:00~10:15, 10:30~11:30, 13:30~15:00, 21:00 ~1:00
NIGHT_MARKET_SQ2 = {'CU': 10, 'PB': 5, 'AL': 5, 'ZN': 5, 'WR': 1, 'NI': 10}
# 上期所夜盘，9:00~10:15, 10:30~11:30, 13:30~15:00, 21:00 ~23:00
NIGHT_MARKET_SQ3 = {'RU': 5, 'RB': 1, 'HC': 1, 'SP': 2, 'FU': 1, 'BU': 2, 'NR': 5, 'C': 1, 'CS': 1}
# 郑商所夜盘，9:00~10:15, 10:30~11:30, 13:30~15:00, 21:00 ~23:00
NIGHT_MARKET_ZZ = {'TA': 2, 'JR': 1, 'OI': 0, 'RO': 1, 'PM': 1, 'WH': 1, 'CF': 5, 'SR': 0, 'FG': 1,
'MA': 1, 'RS': 1, 'RM': 1, 'RI': 1, 'ZC': 0.2}
# 大商所夜盘，9:00~10:15, 10:30~11:30, 13:30~15:00, 21:00 ~23:00
NIGHT_MARKET_DL = {'V': 5, 'L': 5, 'BB': 0.05, 'I': 0.5, 'FB': 0.05, 'C': 1, 'PP': 1, 'A': 1, 'B': 1, 'M': 1, 'Y': 2,
'P': 2,
'JM': 0.5, 'J': 0.5, 'EG': 1}
# 中金日盘，9:15 ~11:30, 13:00~15:15
MARKET_ZJ = {'IC': 0.2, 'IF': 0.2, 'IH': 0.2, 'T': 0.005, 'TF': 0.005, 'TS': 0.005}
# 只有日盘得合约
MARKET_DAY_ONLY = {'IC': 0.2, 'IF': 0.2, 'IH': 0.2, 'T': 0.005, 'TF': 0.005, 'TS': 0.005,
'JD': 1, 'BB': 0.05, 'CS': 1, 'FB': 0.05, 'L': 5, 'V': 5,
'JR': 1, 'LR': 1, 'PM': 1, 'RI': 1, 'RS': 1, 'SM': 2, 'WH': 1, 'AP': 1, 'CJ': 1, 'UR': 1}
# 夜盘23:00收盘的合约
NIGHT_MARKET_23 = {**NIGHT_MARKET_DL, **NIGHT_MARKET_ZZ, **NIGHT_MARKET_SQ3}
@dataclass
class StopOrder:
vt_symbol: str
direction: Direction
offset: Offset
price: float
volume: float
stop_orderid: str
strategy_name: str
lock: bool = False
vt_orderids: list = field(default_factory=list)
status: StopOrderStatus = StopOrderStatus.WAITING
EVENT_CTA_LOG = "eCtaLog"
EVENT_CTA_STRATEGY = "eCtaStrategy"
EVENT_CTA_STOPORDER = "eCtaStopOrder"
INTERVAL_DELTA_MAP = {
Interval.MINUTE: timedelta(minutes=1),
Interval.HOUR: timedelta(hours=1),
Interval.DAILY: timedelta(days=1),
}

1004
vnpy/app/cta_strategy_pro/engine.py Normal file
View File

File diff suppressed because it is too large Load Diff

166
vnpy/app/cta_strategy_pro/strategies/turtle_signal_strategy.py Normal file
View File

@ -0,0 +1,166 @@
from vnpy.app.cta_strategy_pro import (
CtaTemplate,
StopOrder,
Direction,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
class TurtleSignalStrategy(CtaTemplate):
""""""
author = "用Python的交易员"
entry_window = 20
exit_window = 10
atr_window = 20
fixed_size = 1
entry_up = 0
entry_down = 0
exit_up = 0
exit_down = 0
atr_value = 0
long_entry = 0
short_entry = 0
long_stop = 0
short_stop = 0
parameters = ["entry_window", "exit_window", "atr_window", "fixed_size"]
variables = ["entry_up", "entry_down", "exit_up", "exit_down", "atr_value"]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
""""""
super(TurtleSignalStrategy, self).__init__(
cta_engine, strategy_name, vt_symbol, setting
)
self.bg = BarGenerator(self.on_bar)
self.am = ArrayManager()
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(20)
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
self.bg.update_tick(tick)
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.cancel_all()
self.am.update_bar(bar)
if not self.am.inited:
return
# Only calculates new entry channel when no position holding
if not self.pos:
self.entry_up, self.entry_down = self.am.donchian(
self.entry_window
)
self.exit_up, self.exit_down = self.am.donchian(self.exit_window)
if not self.pos:
self.atr_value = self.am.atr(self.atr_window)
self.long_entry = 0
self.short_entry = 0
self.long_stop = 0
self.short_stop = 0
self.send_buy_orders(self.entry_up)
self.send_short_orders(self.entry_down)
elif self.pos > 0:
self.send_buy_orders(self.entry_up)
sell_price = max(self.long_stop, self.exit_down)
self.sell(sell_price, abs(self.pos), True)
elif self.pos < 0:
self.send_short_orders(self.entry_down)
cover_price = min(self.short_stop, self.exit_up)
self.cover(cover_price, abs(self.pos), True)
self.put_event()
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
if trade.direction == Direction.LONG:
self.long_entry = trade.price
self.long_stop = self.long_entry - 2 * self.atr_value
else:
self.short_entry = trade.price
self.short_stop = self.short_entry + 2 * self.atr_value
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
pass
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass
def send_buy_orders(self, price):
""""""
t = self.pos / self.fixed_size
if t < 1:
self.buy(price, self.fixed_size, True)
if t < 2:
self.buy(price + self.atr_value * 0.5, self.fixed_size, True)
if t < 3:
self.buy(price + self.atr_value, self.fixed_size, True)
if t < 4:
self.buy(price + self.atr_value * 1.5, self.fixed_size, True)
def send_short_orders(self, price):
""""""
t = self.pos / self.fixed_size
if t > -1:
self.short(price, self.fixed_size, True)
if t > -2:
self.short(price - self.atr_value * 0.5, self.fixed_size, True)
if t > -3:
self.short(price - self.atr_value, self.fixed_size, True)
if t > -4:
self.short(price - self.atr_value * 1.5, self.fixed_size, True)

451
vnpy/app/cta_strategy_pro/template.py Normal file
View File

@ -0,0 +1,451 @@
""""""
import sys
from abc import ABC
from copy import copy
from typing import Any, Callable
from logging import INFO, ERROR
from vnpy.trader.constant import Interval, Direction, Offset
from vnpy.trader.object import BarData, TickData, OrderData, TradeData
from vnpy.trader.utility import virtual
from .base import StopOrder, EngineType
class CtaComponent(ABC):
""" CTA策略基础组件"""
def __init__(self, strategy=None, **kwargs):
"""
构造
:param strategy:
"""
self.strategy = strategy
# ----------------------------------------------------------------------
def write_log(self, content: str):
"""记录日志"""
if self.strategy:
self.strategy.write_log(msg=content, level=INFO)
else:
print(content)
# ----------------------------------------------------------------------
def write_error(self, content: str, level: int = ERROR):
"""记录错误日志"""
if self.strategy:
self.strategy.write_log(msg=content, level=level)
else:
print(content, file=sys.stderr)
class CtaTemplate(ABC):
"""CTA策略模板"""
author = ""
parameters = []
variables = []
def __init__(
self,
cta_engine: Any,
strategy_name: str,
vt_symbol: str,
setting: dict,
):
""""""
self.cta_engine = cta_engine
self.strategy_name = strategy_name
self.vt_symbol = vt_symbol
self.inited = False
self.trading = False
self.pos = 0
# Copy a new variables list here to avoid duplicate insert when multiple
# strategy instances are created with the same strategy class.
self.variables = copy(self.variables)
self.variables.insert(0, "inited")
self.variables.insert(1, "trading")
self.variables.insert(2, "pos")
self.update_setting(setting)
def update_setting(self, setting: dict):
"""
Update strategy parameter wtih value in setting dict.
"""
for name in self.parameters:
if name in setting:
setattr(self, name, setting[name])
@classmethod
def get_class_parameters(cls):
"""
Get default parameters dict of strategy class.
"""
class_parameters = {}
for name in cls.parameters:
class_parameters[name] = getattr(cls, name)
return class_parameters
def get_parameters(self):
"""
Get strategy parameters dict.
"""
strategy_parameters = {}
for name in self.parameters:
strategy_parameters[name] = getattr(self, name)
return strategy_parameters
def get_variables(self):
"""
Get strategy variables dict.
"""
strategy_variables = {}
for name in self.variables:
strategy_variables[name] = getattr(self, name)
return strategy_variables
def get_data(self):
"""
Get strategy data.
"""
strategy_data = {
"strategy_name": self.strategy_name,
"vt_symbol": self.vt_symbol,
"class_name": self.__class__.__name__,
"author": self.author,
"parameters": self.get_parameters(),
"variables": self.get_variables(),
}
return strategy_data
@virtual
def on_init(self):
"""
Callback when strategy is inited.
"""
pass
@virtual
def on_start(self):
"""
Callback when strategy is started.
"""
pass
@virtual
def on_stop(self):
"""
Callback when strategy is stopped.
"""
pass
@virtual
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
pass
@virtual
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
pass
@virtual
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
pass
@virtual
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
pass
@virtual
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass
def buy(self, price: float, volume: float, stop: bool = False, lock: bool = False, vt_symbol: str = ''):
"""
Send buy order to open a long position.
"""
return self.send_order(vt_symbol=vt_symbol,
direction=Direction.LONG,
offset=Offset.OPEN,
price=price,
volume=volume,
stop=stop,
lock=lock)
def sell(self, price: float, volume: float, stop: bool = False, lock: bool = False, vt_symbol: str = ''):
"""
Send sell order to close a long position.
"""
return self.send_order(vt_symbol=vt_symbol,
direction=Direction.SHORT,
offset=Offset.CLOSE,
price=price,
volume=volume,
stop=stop,
lock=lock)
def short(self, price: float, volume: float, stop: bool = False, lock: bool = False, vt_symbol: str = ''):
"""
Send short order to open as short position.
"""
return self.send_order(vt_symbol=vt_symbol,
direction=Direction.SHORT,
offset=Offset.OPEN,
price=price,
volume=volume,
stop=stop,
lock=lock)
def cover(self, price: float, volume: float, stop: bool = False, lock: bool = False, vt_symbol: str = ''):
"""
Send cover order to close a short position.
"""
return self.send_order(vt_symbol=vt_symbol,
direction=Direction.LONG,
offset=Offset.CLOSE,
price=price,
volume=volume,
stop=stop,
lock=lock)
def send_order(
self,
vt_symbol: str,
direction: Direction,
offset: Offset,
price: float,
volume: float,
stop: bool = False,
lock: bool = False
):
"""
Send a new order.
"""
# 兼容cta_strategy的模板，缺省不指定vt_symbol时，使用策略配置的vt_symbol
if vt_symbol == '':
vt_symbol = self.vt_symbol
if self.trading:
vt_orderids = self.cta_engine.send_order(
self, vt_symbol, direction, offset, price, volume, stop, lock
)
return vt_orderids
else:
return []
def cancel_order(self, vt_orderid: str):
"""
Cancel an existing order.
"""
if self.trading:
self.cta_engine.cancel_order(self, vt_orderid)
def cancel_all(self):
"""
Cancel all orders sent by strategy.
"""
if self.trading:
self.cta_engine.cancel_all(self)
def write_log(self, msg: str, level: int = INFO):
"""
Write a log message.
"""
self.cta_engine.write_log(msg=msg, strategy_name=self.strategy_name, level=level)
def get_engine_type(self):
"""
Return whether the cta_engine is backtesting or live trading.
"""
return self.cta_engine.get_engine_type()
def load_bar(
self,
days: int,
interval: Interval = Interval.MINUTE,
callback: Callable = None,
):
"""
Load historical bar data for initializing strategy.
"""
if not callback:
callback = self.on_bar
self.cta_engine.load_bar(self.vt_symbol, days, interval, callback)
def load_tick(self, days: int):
"""
Load historical tick data for initializing strategy.
"""
self.cta_engine.load_tick(self.vt_symbol, days, self.on_tick)
def put_event(self):
"""
Put an strategy data event for ui update.
"""
if self.inited:
self.cta_engine.put_strategy_event(self)
def send_email(self, msg):
"""
Send email to default receiver.
"""
if self.inited:
self.cta_engine.send_email(msg, self)
def sync_data(self):
"""
Sync strategy variables value into disk storage.
"""
if self.trading:
self.cta_engine.sync_strategy_data(self)
class CtaSignal(ABC):
""""""
def __init__(self):
""""""
self.signal_pos = 0
@virtual
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
pass
@virtual
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
pass
def set_signal_pos(self, pos):
""""""
self.signal_pos = pos
def get_signal_pos(self):
""""""
return self.signal_pos
class TargetPosTemplate(CtaTemplate):
""""""
tick_add = 1
last_tick = None
last_bar = None
target_pos = 0
vt_orderids = []
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
""""""
super(TargetPosTemplate, self).__init__(
cta_engine, strategy_name, vt_symbol, setting
)
self.variables.append("target_pos")
@virtual
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
self.last_tick = tick
if self.trading:
self.trade()
@virtual
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.last_bar = bar
@virtual
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
vt_orderid = order.vt_orderid
if not order.is_active() and vt_orderid in self.vt_orderids:
self.vt_orderids.remove(vt_orderid)
def set_target_pos(self, target_pos):
""""""
self.target_pos = target_pos
self.trade()
def trade(self):
""""""
self.cancel_all()
pos_change = self.target_pos - self.pos
if not pos_change:
return
long_price = 0
short_price = 0
if self.last_tick:
if pos_change > 0:
long_price = self.last_tick.ask_price_1 + self.tick_add
if self.last_tick.limit_up:
long_price = min(long_price, self.last_tick.limit_up)
else:
short_price = self.last_tick.bid_price_1 - self.tick_add
if self.last_tick.limit_down:
short_price = max(short_price, self.last_tick.limit_down)
else:
if pos_change > 0:
long_price = self.last_bar.close_price + self.tick_add
else:
short_price = self.last_bar.close_price - self.tick_add
if self.get_engine_type() == EngineType.BACKTESTING:
if pos_change > 0:
vt_orderids = self.buy(long_price, abs(pos_change))
else:
vt_orderids = self.short(short_price, abs(pos_change))
self.vt_orderids.extend(vt_orderids)
else:
if self.vt_orderids:
return
if pos_change > 0:
if self.pos < 0:
if pos_change < abs(self.pos):
vt_orderids = self.cover(long_price, pos_change)
else:
vt_orderids = self.cover(long_price, abs(self.pos))
else:
vt_orderids = self.buy(long_price, abs(pos_change))
else:
if self.pos > 0:
if abs(pos_change) < self.pos:
vt_orderids = self.sell(short_price, abs(pos_change))
else:
vt_orderids = self.sell(short_price, abs(self.pos))
else:
vt_orderids = self.short(short_price, abs(pos_change))
self.vt_orderids.extend(vt_orderids)

1
vnpy/app/cta_strategy_pro/ui/__init__.py Normal file
View File

@ -0,0 +1 @@
from .widget import CtaManager

BIN
vnpy/app/cta_strategy_pro/ui/cta.ico Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 66 KiB

455
vnpy/app/cta_strategy_pro/ui/widget.py Normal file
View File

@ -0,0 +1,455 @@
from vnpy.event import Event, EventEngine
from vnpy.trader.engine import MainEngine
from vnpy.trader.ui import QtCore, QtGui, QtWidgets
from vnpy.trader.ui.widget import (
BaseCell,
EnumCell,
MsgCell,
TimeCell,
BaseMonitor
)
from ..base import (
APP_NAME,
EVENT_CTA_LOG,
EVENT_CTA_STOPORDER,
EVENT_CTA_STRATEGY
)
from ..engine import CtaEngine
class CtaManager(QtWidgets.QWidget):
""""""
signal_log = QtCore.pyqtSignal(Event)
signal_strategy = QtCore.pyqtSignal(Event)
def __init__(self, main_engine: MainEngine, event_engine: EventEngine):
super(CtaManager, self).__init__()
self.main_engine = main_engine
self.event_engine = event_engine
self.cta_engine = main_engine.get_engine(APP_NAME)
self.managers = {}
self.init_ui()
self.register_event()
self.cta_engine.init_engine()
self.update_class_combo()
def init_ui(self):
""""""
self.setWindowTitle("CTA策略")
# Create widgets
self.class_combo = QtWidgets.QComboBox()
add_button = QtWidgets.QPushButton("添加策略")
add_button.clicked.connect(self.add_strategy)
init_button = QtWidgets.QPushButton("全部初始化")
init_button.clicked.connect(self.cta_engine.init_all_strategies)
start_button = QtWidgets.QPushButton("全部启动")
start_button.clicked.connect(self.cta_engine.start_all_strategies)
stop_button = QtWidgets.QPushButton("全部停止")
stop_button.clicked.connect(self.cta_engine.stop_all_strategies)
clear_button = QtWidgets.QPushButton("清空日志")
clear_button.clicked.connect(self.clear_log)
self.scroll_layout = QtWidgets.QVBoxLayout()
self.scroll_layout.addStretch()
scroll_widget = QtWidgets.QWidget()
scroll_widget.setLayout(self.scroll_layout)
scroll_area = QtWidgets.QScrollArea()
scroll_area.setWidgetResizable(True)
scroll_area.setWidget(scroll_widget)
self.log_monitor = LogMonitor(self.main_engine, self.event_engine)
self.stop_order_monitor = StopOrderMonitor(
self.main_engine, self.event_engine
)
# Set layout
hbox1 = QtWidgets.QHBoxLayout()
hbox1.addWidget(self.class_combo)
hbox1.addWidget(add_button)
hbox1.addStretch()
hbox1.addWidget(init_button)
hbox1.addWidget(start_button)
hbox1.addWidget(stop_button)
hbox1.addWidget(clear_button)
grid = QtWidgets.QGridLayout()
grid.addWidget(scroll_area, 0, 0, 2, 1)
grid.addWidget(self.stop_order_monitor, 0, 1)
grid.addWidget(self.log_monitor, 1, 1)
vbox = QtWidgets.QVBoxLayout()
vbox.addLayout(hbox1)
vbox.addLayout(grid)
self.setLayout(vbox)
def update_class_combo(self):
""""""
self.class_combo.addItems(
self.cta_engine.get_all_strategy_class_names()
)
def register_event(self):
""""""
self.signal_strategy.connect(self.process_strategy_event)
self.event_engine.register(
EVENT_CTA_STRATEGY, self.signal_strategy.emit
)
def process_strategy_event(self, event):
"""
Update strategy status onto its monitor.
"""
data = event.data
strategy_name = data["strategy_name"]
if strategy_name in self.managers:
manager = self.managers[strategy_name]
manager.update_data(data)
else:
manager = StrategyManager(self, self.cta_engine, data)
self.scroll_layout.insertWidget(0, manager)
self.managers[strategy_name] = manager
def remove_strategy(self, strategy_name):
""""""
manager = self.managers.pop(strategy_name)
manager.deleteLater()
def add_strategy(self):
""""""
class_name = str(self.class_combo.currentText())
if not class_name:
return
parameters = self.cta_engine.get_strategy_class_parameters(class_name)
editor = SettingEditor(parameters, class_name=class_name)
n = editor.exec_()
if n == editor.Accepted:
setting = editor.get_setting()
vt_symbol = setting.pop("vt_symbol")
strategy_name = setting.pop("strategy_name")
self.cta_engine.add_strategy(
class_name, strategy_name, vt_symbol, setting
)
def clear_log(self):
""""""
self.log_monitor.setRowCount(0)
def show(self):
""""""
self.showMaximized()
class StrategyManager(QtWidgets.QFrame):
"""
Manager for a strategy
"""
def __init__(
self, cta_manager: CtaManager, cta_engine: CtaEngine, data: dict
):
""""""
super(StrategyManager, self).__init__()
self.cta_manager = cta_manager
self.cta_engine = cta_engine
self.strategy_name = data["strategy_name"]
self._data = data
self.init_ui()
def init_ui(self):
""""""
self.setFixedHeight(300)
self.setFrameShape(self.Box)
self.setLineWidth(1)
init_button = QtWidgets.QPushButton("初始化")
init_button.clicked.connect(self.init_strategy)
start_button = QtWidgets.QPushButton("启动")
start_button.clicked.connect(self.start_strategy)
stop_button = QtWidgets.QPushButton("停止")
stop_button.clicked.connect(self.stop_strategy)
edit_button = QtWidgets.QPushButton("编辑")
edit_button.clicked.connect(self.edit_strategy)
remove_button = QtWidgets.QPushButton("移除")
remove_button.clicked.connect(self.remove_strategy)
reload_button = QtWidgets.QPushButton("重载")
reload_button.clicked.connect(self.reload_strategy)
strategy_name = self._data["strategy_name"]
vt_symbol = self._data["vt_symbol"]
class_name = self._data["class_name"]
author = self._data["author"]
label_text = (
f"{strategy_name} - {vt_symbol} ({class_name} by {author})"
)
label = QtWidgets.QLabel(label_text)
label.setAlignment(QtCore.Qt.AlignCenter)
self.parameters_monitor = DataMonitor(self._data["parameters"])
self.variables_monitor = DataMonitor(self._data["variables"])
hbox = QtWidgets.QHBoxLayout()
hbox.addWidget(init_button)
hbox.addWidget(start_button)
hbox.addWidget(stop_button)
hbox.addWidget(edit_button)
hbox.addWidget(remove_button)
hbox.addWidget(reload_button)
vbox = QtWidgets.QVBoxLayout()
vbox.addWidget(label)
vbox.addLayout(hbox)
vbox.addWidget(self.parameters_monitor)
vbox.addWidget(self.variables_monitor)
self.setLayout(vbox)
def update_data(self, data: dict):
""""""
self._data = data
self.parameters_monitor.update_data(data["parameters"])
self.variables_monitor.update_data(data["variables"])
def init_strategy(self):
""""""
self.cta_engine.init_strategy(self.strategy_name)
def start_strategy(self):
""""""
self.cta_engine.start_strategy(self.strategy_name)
def stop_strategy(self):
""""""
self.cta_engine.stop_strategy(self.strategy_name)
def edit_strategy(self):
""""""
strategy_name = self._data["strategy_name"]
parameters = self.cta_engine.get_strategy_parameters(strategy_name)
editor = SettingEditor(parameters, strategy_name=strategy_name)
n = editor.exec_()
if n == editor.Accepted:
setting = editor.get_setting()
self.cta_engine.edit_strategy(strategy_name, setting)
def remove_strategy(self):
""""""
result = self.cta_engine.remove_strategy(self.strategy_name)
# Only remove strategy gui manager if it has been removed from engine
if result:
self.cta_manager.remove_strategy(self.strategy_name)
def reload_strategy(self):
"""重新加载策略"""
self.cta_engine.reload_strategy(self.strategy_name)
class DataMonitor(QtWidgets.QTableWidget):
"""
Table monitor for parameters and variables.
"""
def __init__(self, data: dict):
""""""
super(DataMonitor, self).__init__()
self._data = data
self.cells = {}
self.init_ui()
def init_ui(self):
""""""
labels = list(self._data.keys())
self.setColumnCount(len(labels))
self.setHorizontalHeaderLabels(labels)
self.setRowCount(1)
self.verticalHeader().setSectionResizeMode(
QtWidgets.QHeaderView.Stretch
)
self.verticalHeader().setVisible(False)
self.setEditTriggers(self.NoEditTriggers)
for column, name in enumerate(self._data.keys()):
value = self._data[name]
cell = QtWidgets.QTableWidgetItem(str(value))
cell.setTextAlignment(QtCore.Qt.AlignCenter)
self.setItem(0, column, cell)
self.cells[name] = cell
def update_data(self, data: dict):
""""""
for name, value in data.items():
cell = self.cells[name]
cell.setText(str(value))
class StopOrderMonitor(BaseMonitor):
"""
Monitor for local stop order.
"""
event_type = EVENT_CTA_STOPORDER
data_key = "stop_orderid"
sorting = True
headers = {
"stop_orderid": {
"display": "停止委托号",
"cell": BaseCell,
"update": False,
},
"vt_orderids": {"display": "限价委托号", "cell": BaseCell, "update": True},
"vt_symbol": {"display": "本地代码", "cell": BaseCell, "update": False},
"direction": {"display": "方向", "cell": EnumCell, "update": False},
"offset": {"display": "开平", "cell": EnumCell, "update": False},
"price": {"display": "价格", "cell": BaseCell, "update": False},
"volume": {"display": "数量", "cell": BaseCell, "update": False},
"status": {"display": "状态", "cell": EnumCell, "update": True},
"lock": {"display": "锁仓", "cell": BaseCell, "update": False},
"strategy_name": {"display": "策略名", "cell": BaseCell, "update": False},
}
class LogMonitor(BaseMonitor):
"""
Monitor for log data.
"""
event_type = EVENT_CTA_LOG
data_key = ""
sorting = False
headers = {
"time": {"display": "时间", "cell": TimeCell, "update": False},
"msg": {"display": "信息", "cell": MsgCell, "update": False},
}
def init_ui(self):
"""
Stretch last column.
"""
super(LogMonitor, self).init_ui()
self.horizontalHeader().setSectionResizeMode(
1, QtWidgets.QHeaderView.Stretch
)
def insert_new_row(self, data):
"""
Insert a new row at the top of table.
"""
super(LogMonitor, self).insert_new_row(data)
self.resizeRowToContents(0)
class SettingEditor(QtWidgets.QDialog):
"""
For creating new strategy and editing strategy parameters.
"""
def __init__(
self, parameters: dict, strategy_name: str = "", class_name: str = ""
):
""""""
super(SettingEditor, self).__init__()
self.parameters = parameters
self.strategy_name = strategy_name
self.class_name = class_name
self.edits = {}
self.init_ui()
def init_ui(self):
""""""
form = QtWidgets.QFormLayout()
# Add vt_symbol and name edit if add new strategy
if self.class_name:
self.setWindowTitle(f"添加策略：{self.class_name}")
button_text = "添加"
parameters = {"strategy_name": "", "vt_symbol": ""}
parameters.update(self.parameters)
else:
self.setWindowTitle(f"参数编辑：{self.strategy_name}")
button_text = "确定"
parameters = self.parameters
for name, value in parameters.items():
type_ = type(value)
edit = QtWidgets.QLineEdit(str(value))
if type_ is int:
validator = QtGui.QIntValidator()
edit.setValidator(validator)
elif type_ is float:
validator = QtGui.QDoubleValidator()
edit.setValidator(validator)
form.addRow(f"{name} {type_}", edit)
self.edits[name] = (edit, type_)
button = QtWidgets.QPushButton(button_text)
button.clicked.connect(self.accept)
form.addRow(button)
self.setLayout(form)
def get_setting(self):
""""""
setting = {}
if self.class_name:
setting["class_name"] = self.class_name
for name, tp in self.edits.items():
edit, type_ = tp
value_text = edit.text()
if type_ == bool:
if value_text == "True":
value = True
else:
value = False
else:
value = type_(value_text)
setting[name] = value
return setting