cfdaily
277 lines
10 KiB
Python
277 lines
10 KiB
Python
"""
|
||
纯突破量化策略
|
||
逻辑: N日新高放量突破买入，严格止损止盈卖出
|
||
"""
|
||
import pandas as pd
|
||
import numpy as np
|
||
from typing import Dict, List, Optional
|
||
|
||
from vnpy.trader.app.ctaStrategy import CtaTemplate
|
||
from vnpy.trader.object import BarData, TickData
|
||
from vnpy.trader.constant import Direction
|
||
|
||
|
||
class PureBreakoutStrategy(CtaTemplate):
|
||
"""
|
||
纯突破策略
|
||
- 突破N日新高 + 放量买入
|
||
- 严格止损（跌破突破最低价5%）
|
||
- 跟踪止盈（从高点回落10%）
|
||
- 均线止盈（跌破均线）
|
||
- 最长持有到期卖出
|
||
- 等权配置，单票不超过5%仓位
|
||
"""
|
||
|
||
# 策略参数
|
||
author = "翼德"
|
||
parameters = [
|
||
"breakout_days", # 突破N日新高
|
||
"volume_multiple", # 放量倍数
|
||
"stop_loss_pct", # 止损百分比（相对于突破最低价）
|
||
"trailing_stop_pct", # 跟踪止盈百分比（从高点回落）
|
||
"ma_period", # 均线周期
|
||
"max_holding_days", # 最长持有天数
|
||
"max_position_pct", # 单票最大仓位百分比
|
||
]
|
||
|
||
# 策略变量
|
||
variables = [
|
||
"holdings", # 当前持仓信息字典
|
||
]
|
||
|
||
def __init__(self, cta_engine, strategy_name, setting_dict):
|
||
super().__init__(cta_engine, strategy_name, setting_dict)
|
||
|
||
# 默认参数
|
||
self.breakout_days = getattr(self, 'breakout_days', 60)
|
||
self.volume_multiple = getattr(self, 'volume_multiple', 1.5)
|
||
self.stop_loss_pct = getattr(self, 'stop_loss_pct', 0.05)
|
||
self.trailing_stop_pct = getattr(self, 'trailing_stop_pct', 0.10)
|
||
self.ma_period = getattr(self, 'ma_period', 20)
|
||
self.max_holding_days = getattr(self, 'max_holding_days', 60)
|
||
self.max_position_pct = getattr(self, 'max_position_pct', 0.05)
|
||
|
||
# 持仓信息存储
|
||
# {
|
||
# symbol: {
|
||
# 'entry_date': datetime, # 买入日期
|
||
# 'entry_price': float, # 买入价格
|
||
# 'breakout_low': float, # 突破日最低价（止损用）
|
||
# 'highest_price': float, # 持仓以来最高价（跟踪止盈用）
|
||
# 'holding_days': int, # 持有天数
|
||
# }
|
||
# }
|
||
self.holdings: Dict[str, dict] = {}
|
||
|
||
def on_init(self):
|
||
"""策略初始化"""
|
||
self.write_log(f"纯突破策略初始化，突破天数={self.breakout_days}，单票最大仓位={self.max_position_pct*100:.0f}%")
|
||
self.load_bar(self.breakout_days + self.max_holding_days) # 加载足够数据计算指标
|
||
|
||
def on_start(self):
|
||
"""策略启动"""
|
||
self.write_log("策略启动")
|
||
|
||
def on_stop(self):
|
||
"""策略停止"""
|
||
self.write_log("策略停止")
|
||
|
||
def on_bar(self, bar: BarData):
|
||
"""K线推送，主逻辑"""
|
||
symbol = bar.vt_symbol
|
||
|
||
# 更新持仓天数和最高价
|
||
if symbol in self.holdings:
|
||
self._update_holding(symbol, bar)
|
||
|
||
# 检查卖出条件
|
||
if self._check_exit(symbol, bar):
|
||
self._exit_position(symbol, bar)
|
||
return
|
||
|
||
# 如果还没持仓，检查买入条件
|
||
if symbol not in self.holdings:
|
||
if self._check_entry(symbol, bar):
|
||
self._enter_position(symbol, bar)
|
||
return
|
||
|
||
def _update_holding(self, symbol: str, bar: BarData):
|
||
"""更新持仓信息"""
|
||
hold = self.holdings[symbol]
|
||
hold['holding_days'] += 1
|
||
|
||
# 更新最高价
|
||
if bar.close > hold['highest_price']:
|
||
hold['highest_price'] = bar.close
|
||
|
||
def _check_exit(self, symbol: str, bar: BarData) -> bool:
|
||
"""检查是否满足卖出条件，满足一个就卖出"""
|
||
hold = self.holdings[symbol]
|
||
|
||
# 条件1: 止损 - 跌破突破最低价的(1-stop_loss_pct)
|
||
stop_price = hold['breakout_low'] * (1 - self.stop_loss_pct)
|
||
if bar.low <= stop_price:
|
||
self.write_log(f"{symbol}: 触发止损，价格{bar.low:.2f} <= 止损价{stop_price:.2f}")
|
||
return True
|
||
|
||
# 条件2: 跟踪止盈 - 从最高点回落超过trailing_stop_pct
|
||
trailing_stop = hold['highest_price'] * (1 - self.trailing_stop_pct)
|
||
if bar.close <= trailing_stop:
|
||
self.write_log(f"{symbol}: 触发跟踪止盈，现价{bar.close:.2f} <= 止盈价{trailing_stop:.2f}，最高{hold['highest_price']:.2f}")
|
||
return True
|
||
|
||
# 条件3: 均线止盈 - 收盘价跌破均线
|
||
ma = self.calculate_ma(symbol, self.ma_period)
|
||
if ma is not None and bar.close < ma:
|
||
self.write_log(f"{symbol}: 触发均线止盈，现价{bar.close:.2f} < MA{self.ma_period}={ma:.2f}")
|
||
return True
|
||
|
||
# 条件4: 最长持有天数到期
|
||
if hold['holding_days'] >= self.max_holding_days:
|
||
self.write_log(f"{symbol}: 持有到期{self.max_holding_days}天，卖出")
|
||
return True
|
||
|
||
# 都不满足，继续持有
|
||
return False
|
||
|
||
def _check_entry(self, symbol: str, bar: BarData) -> bool:
|
||
"""检查是否满足买入条件"""
|
||
# 检查是否已经达到单票最大仓位限制（整个策略层面）
|
||
current_total = self.calculate_current_total_position()
|
||
if current_total >= self.balance * self.max_position_pct:
|
||
# 已经有仓位达到上限，不开新仓
|
||
return False
|
||
|
||
# 条件1: 收盘价创N日新高
|
||
if not self._is_new_high(symbol, bar.close):
|
||
return False
|
||
|
||
# 条件2: 成交量放量相对于N日均量
|
||
if not self._is_volume_enough(symbol, bar.volume):
|
||
return False
|
||
|
||
# 两个条件都满足，可以买入
|
||
self.write_log(f"{symbol}: 满足突破条件，N={self.breakout_days}日新高，放量")
|
||
return True
|
||
|
||
def _is_new_high(self, symbol: str, current_close: float) -> bool:
|
||
"""判断是否创N日新高"""
|
||
# 获取最近N天的收盘价（不包含今天）
|
||
bars = self.get_bars(symbol, self.breakout_days)
|
||
if len(bars) < self.breakout_days:
|
||
return False # 数据不够，不操作
|
||
|
||
# 计算N日最高收盘价
|
||
highest = max(b.high for b in bars)
|
||
|
||
# 当前收盘价创新高
|
||
return current_close > highest
|
||
|
||
def _is_volume_enough(self, symbol: str, current_volume: float) -> bool:
|
||
"""判断成交量是否足够放量"""
|
||
# 获取最近N天的成交量
|
||
bars = self.get_bars(symbol, self.breakout_days)
|
||
if len(bars) < self.breakout_days:
|
||
return False
|
||
|
||
# 计算平均成交量
|
||
avg_volume = np.mean([b.volume for b in bars])
|
||
|
||
# 当前成交量大于等于倍数 * 平均
|
||
return current_volume >= avg_volume * self.volume_multiple
|
||
|
||
def calculate_ma(self, symbol: str, period: int) -> Optional[float]:
|
||
"""计算均线"""
|
||
bars = self.get_bars(symbol, period)
|
||
if len(bars) < period:
|
||
return None
|
||
|
||
close_list = [b.close for b in bars]
|
||
return np.mean(close_list)
|
||
|
||
def calculate_current_total_position(self) -> float:
|
||
"""计算当前总持仓市值"""
|
||
total = 0.0
|
||
for symbol, hold in self.holdings.items():
|
||
# 获取最新价格
|
||
last_bar = self.get_last_bar(symbol)
|
||
if last_bar:
|
||
# 持仓数量 * 最新价格
|
||
position = self.get_position(symbol)
|
||
if position:
|
||
total += abs(position.volume) * last_bar.close
|
||
|
||
return total
|
||
|
||
def _enter_position(self, symbol: str, bar: BarData):
|
||
"""开仓买入"""
|
||
# 计算目标仓位: 最大仓位对应市值
|
||
target_value = self.balance * self.max_position_pct
|
||
|
||
# 计算买入股数（A股100股整数）
|
||
price = bar.open # 突破次日开盘买入，如果是当日就是开盘
|
||
if price <= 0:
|
||
price = bar.close
|
||
|
||
target_volume = int(target_value / price / 100) * 100
|
||
|
||
if target_volume <= 0:
|
||
self.write_log(f"{symbol}: 计算得到目标股数{target_volume}，不买入")
|
||
return
|
||
|
||
# 买入
|
||
self.buy(symbol, price, target_volume)
|
||
|
||
# 记录持仓信息
|
||
self.holdings[symbol] = {
|
||
'entry_date': bar.datetime,
|
||
'entry_price': price,
|
||
'breakout_low': bar.low, # 突破日最低价作为止损基准
|
||
'highest_price': bar.close, # 最高价从买入后开始计算
|
||
'holding_days': 0, # 买入当日算0天
|
||
}
|
||
|
||
self.write_log(f"{symbol}: 买入成功，数量{target_volume}，价格{price:.2f}，止损{bar.low*(1-self.stop_loss_pct):.2f}")
|
||
|
||
self.put_order()
|
||
|
||
def _exit_position(self, symbol: str, bar: BarData):
|
||
"""平仓卖出"""
|
||
position = self.get_position(symbol)
|
||
|
||
if position is None or position.volume <= 0:
|
||
del self.holdings[symbol]
|
||
return
|
||
|
||
# 全部卖出
|
||
price = bar.close
|
||
self.sell(symbol, price, abs(position.volume))
|
||
|
||
# 计算盈亏
|
||
hold = self.holdings[symbol]
|
||
pnl = (price - hold['entry_price']) * position.volume
|
||
pct = (price - hold['entry_price']) / hold['entry_price'] * 100
|
||
|
||
self.write_log(f"{symbol}: 卖出成功，盈亏{pnl:.2f} ({pct:.2f}%)，持有{hold['holding_days']}天")
|
||
|
||
# 从持仓记录删除
|
||
del self.holdings[symbol]
|
||
|
||
self.put_order()
|
||
|
||
def get_bars(self, symbol: str, days: int):
|
||
"""获取最近N根Bar（不包含当前Bar）"""
|
||
# 这个接口根据vn.py实际使用调整，这里给出标准接口
|
||
return self.cta_engine.get_bars(symbol, days)
|
||
|
||
def get_last_bar(self, symbol: str):
|
||
"""获取最新一根Bar"""
|
||
bars = self.get_bars(symbol, 1)
|
||
if bars:
|
||
return bars[-1]
|
||
return None
|
||
|
||
def get_position(self, symbol: str):
|
||
"""获取当前持仓"""
|
||
return self.cta_engine.get_position(symbol)
|