闲来无事,在家把之前的编写的策略如何一步一步实现的记录下来了!
在B站发了些自己做的关于量化的视频,只是简单的记录一下自己是
如何一步一步走进这个量化坑的,希望可以帮助新手少走一些坑。
这个策略,实盘肯定不行的,只是记录下编写策略的一个流程。
我在B站发的视频,欢迎大家点赞关注 /😊/😊/😊/😊/😊/😊/😊
https://www.bilibili.com/video/BV1pV411r7Kt?p=10
视频只是简单的记录自己之前做的一些思路,高手别喷(●ˇ∀ˇ●)
质量肯定没法和官方的比 最后还是希望大家多多支持官方
from vnpy.app.cta_strategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
from vnpy.trader.constant import Interval
class Demo01(CtaTemplate):
""""""
# 参数
fast_window = 30 # 快速均线
slow_window = 60 # 慢速均线
x_min = 30 # 交易周期 默认是15分钟
lots = 1 # 开仓手数
save = 20 # 止损参数
startstop = 100 # 开始止盈
stoploss = 40 # 回撤点位
daily_window = 10 # 日线的参数
max_lots = 3
flag=0
# 变量
fast_ma = 0
fast_ma_pre = 0
slow_ma = 0
slow_ma_pre = 0
price = 0 # tick的实时价格
bartime = "" # 时间的显示
avg_buy_price = 0
avg_sell_price = 0
highest = 0
lowest = 0
daily_ma = 0
liqDays = 60
liqpoint = 0
holding_days = 0
run_buy=False
run_sell=False
parameters = ["fast_window", "slow_window", "x_min", "lots", 'stoploss', 'startstop', 'save']
variables = ["bartime", "price", "fast_ma", "slow_ma", 'highest', 'lowest']
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
""""""
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.bg_x = BarGenerator(self.on_bar, self.x_min, self.on_x_bar, Interval.MINUTE)
self.am_x = ArrayManager()
self.bg_daily = BarGenerator(self.on_bar, 1, self.on_daily_bar, Interval.DAILY)
self.am_daliy = ArrayManager()
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(30)
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_x.update_tick(tick)
self.price = tick.last_price
if self.pos > 0:
if tick.last_price < self.avg_buy_price - self.save*2:
self.sell(tick.last_price*0.9, abs(self.pos))
print(tick.datetime, tick.last_price, "保命出场")
self.flag+=1
elif self.pos < 0:
if tick.last_price > self.avg_sell_price + self.save*2:
self.cover(tick.last_price*1.2, abs(self.pos))
print(tick.datetime, tick.last_price, "保命出场")
self.flag+=1
self.put_event()
def on_bar(self, bar: BarData):
self.bg_x.update_bar(bar)
self.bg_daily.update_bar(bar)
def on_x_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.cancel_all()
am = self.am_x
am.update_bar(bar)
if not am.inited:
return
# ma30的计算
self.fast_ma = am.close_array[-self.fast_window:-1].mean()
# 之前的ma30的计算
self.fast_ma_pre = am.close_array[-self.fast_window - 1:-2].mean()
# ma60的计算
self.slow_ma = am.close_array[-self.slow_window:-1].mean()
# 之前的ma60的计算
self.slow_ma_pre = am.close_array[-self.slow_window - 1:-2].mean()
self.run_buy=bar.close_price<=am.close_array[-2]*1.05
self.run_sell=bar.close_price>=am.close_array[-2]*0.95
if self.pos == 0:
# 金叉开多
if self.fast_ma_pre < self.slow_ma_pre and self.fast_ma > self.slow_ma:
if bar.close_price > self.daily_ma and self.run_buy:
self.buy(bar.close_price,min(self.lots,self.max_lots))
self.highest = bar.close_price
self.avg_buy_price = bar.close_price
self.holding_days = 0
print(bar.datetime, bar.close_price, "开多单","开仓手数:",min(self.lots,self.max_lots))
elif self.fast_ma_pre > self.slow_ma_pre and self.fast_ma < self.slow_ma:
# 死叉开空
if bar.close_price < self.daily_ma and self.run_sell:
self.short(bar.close_price,min(self.lots,self.max_lots))
self.lowest = bar.close_price
self.avg_sell_price = bar.close_price
self.holding_days = 0
print(bar.datetime, bar.close_price, "开空单","开仓手数:",min(self.lots,self.max_lots))
elif self.pos > 0:
# 持有多单的最高价记录
self.highest = max(bar.high_price, self.highest)
# 多单止损的逻辑
if bar.close_price < self.avg_buy_price - self.save:
self.sell(bar.close_price, abs(self.pos))
print(bar.datetime, bar.close_price, "多单止损","止损手数:",abs(self.pos))
self.flag+=1
# 多单止盈
elif self.highest > self.avg_buy_price + self.startstop:
if bar.close_price < self.highest - self.stoploss:
self.sell(bar.close_price, abs(self.pos))
print(bar.datetime, bar.close_price, "多单止盈","止盈手数:",abs(self.pos))
self.flag=0
elif self.holding_days > 20 and bar.close_price < self.liqpoint:
self.sell(bar.close_price, abs(self.pos))
print(bar.datetime, bar.close_price, "多单自适应均线出场","平仓手数:",abs(self.pos))
# if self.avg_buy_price<bar.close_price:
# self.flag=0
# else:
# self.flag+=1
elif self.pos < 0:
# 持有空单的最低价记录
self.lowest = min(bar.low_price, self.lowest)
# 空单止损的逻辑
if bar.close_price > self.avg_sell_price + self.save:
self.cover(bar.close_price, abs(self.pos))
print(bar.datetime, bar.close_price, "空单止损","止损手数:",abs(self.pos))
self.flag+=1
# 空单止盈
elif self.lowest < self.avg_sell_price - self.startstop:
if bar.close_price > self.lowest + self.stoploss:
self.cover(bar.close_price, abs(self.pos))
print(bar.datetime, bar.close_price, "空单止盈", "止盈手数:", abs(self.pos))
self.flag=0
elif self.holding_days > 20 and bar.close_price > self.liqpoint:
self.cover(bar.close_price, abs(self.pos))
print(bar.datetime, bar.close_price, "空单自适应均线出场","平仓手数:",abs(self.pos))
# if self.avg_sell_price > bar.close_price:
# self.flag =0
# else:
# self.flag += 1
if self.pos != 0:
self.holding_days += 1
else:
self.liqDays = self.slow_window
if self.pos != 0 and self.holding_days >= 20:
self.liqDays -= 1
self.liqDays = max(self.liqDays, 50)
self.liqpoint = am.close_array[-self.liqDays:].mean()
self.lots=2 if self.flag>=2 else 1
self.put_event()
def on_daily_bar(self, bar: BarData):
self.cancel_all()
am = self.am_daliy
am.update_bar(bar)
if not am.inited:
return
self.daily_ma = am.close_array[-self.daily_window:-1].mean()
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
pass
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
self.put_event()
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass