聚焦“AI 策略止盈后冷却期同一标的 30 日内不再二次买入”这一件事适合直接写进课程讲义或技术博客。AI 策略止盈后冷却期同一标的 30 日内不再二次买入一、实际应用场景描述在 AI 选股 趋势策略中一个非常常见、但很少被系统化处理的问题是刚止盈卖出的标的没几天又被策略买回来了。典型场景场景 问题本质一波趋势末端止盈卖出 3 天后 AI 模型再次打分选中同一只股票卖出后股价横盘震荡 策略反复买入 → 小亏止损 → 再买入短期情绪驱动的假突破 刚走完一波结构尚未修复二次进场必亏止盈后立刻接回 把已经落袋的利润又还给市场 核心矛盾AI 模型是健忘的 —— 它不知道自己 5 天前刚推荐卖出过这只股票。如果不显式引入冷却期Cooling Period机制策略会在同一个技术性顶部反复挨打。二、引入痛点问题结构化我们把这个问题拆成 5 个可被工程化解决的层级层级 痛点 后果数据层 策略不记录为什么卖出 无法区分止盈和止损后的再买入模型层 AI 模型无状态 每次打分独立不记得历史操作策略层 缺乏冷却期约束 止盈 → 3 天后再买入 → 利润回吐风控层 没有同一标的频率限制 资金被锁在低效交易中教学层 回测忽略再买入问题 实盘才发现纸上富贵本质结论没有冷却期的止盈不是止盈是卖出后看戏 3 天再买回来。三、核心逻辑讲解为什么要这么设计3.1 为什么是 30 日30 天不是一个玄学数字而是来自市场微观结构经验逻辑 说明A 股短期情绪波动周期 通常为 2~4 周技术形态修复时间 均线重新发散、量价重新配合机构资金再配置周期 基金经理调仓通常以月为单位回测实证 10 日太短震荡市频繁触发60 日太长错过二次趋势 30 日是一个在回测中可验证、在实盘中可解释的参数。3.2 冷却期状态机设计我们用一个有限状态机FSM来管理每只股票的冷却状态┌──────────────────────────────────────────────────────────┐│ 标的冷却状态机 │├──────────────────────────────────────────────────────────┤│ ││ ┌─────────┐ 买入 ┌──────────┐ ││ │ 冷却中 │ ──────→ │ 持仓中 │ ││ │ cooldown │ │ holding │ ││ └─────────┘ └────┬─────┘ ││ ▲ │ ││ │ 冷却期结束 │ 止盈卖出 ││ │ ▼ ││ ┌─────────┐ ┌──────────┐ ││ │ 可交易 │ ←────── │ 止盈冷却 │ ││ │ ready │ 30日 │ just_tp │ ││ └─────────┘ └──────────┘ ││ ││ 状态转换事件 ││ • BUY → holding ││ • SELL(TP) → just_tp → 开始 30 日倒计时 ││ • 30 天后 → ready ││ • SELL(SL) → ready止损不进入冷却期 │└──────────────────────────────────────────────────────────┘3.3 核心数据结构设计# 冷却期状态存储self.cooldown_state: Dict[str, Dict] {000001: {status: cooling, # ready / cooling / holdingcooling_until: Timestamp, # 冷却结束日期last_exit_date: Timestamp, # 上次退出日期last_exit_reason: tp, # tp / sl / manualdays_since_exit: 12 # 已冷却天数调试用}}3.4 正确集成到策略中的位置AI 模型打分↓生成候选列表↓★ 冷却期过滤剔除 cooling 状态标的↓执行买入↓持有期间...↓止盈卖出 → 进入 cooling 状态30 天↓止损卖出 → 进入 ready 状态可立即再买入关键设计决策只有止盈Take Profit触发冷却期。止损Stop Loss不触发 —— 因为止损意味着看错模型应该有机会纠错立刻再买。四、项目结构工程化cooling_period/├── README.md├── requirements.txt├── config.yaml├── data/│ └── daily_prices.csv├── src/│ ├── data_loader.py│ ├── signal_generator.py│ ├── cooldown_manager.py # ★ 冷却期核心模块│ ├── strategy_engine.py # 策略引擎集成冷却期│ ├── backtester.py # 回测框架│ └── visualizer.py # 可视化├── main.py└── compare_cooldown.py # 有/无冷却期对比实验五、完整代码模块化 清晰注释requirements.txtpandas1.5numpy1.21matplotlib3.5seaborn0.12scipy1.9pyyaml6.0config.yaml# 止盈冷却期配置# ★ 冷却期参数cooldown:enabled: truecooling_days: 30 # 止盈后冷却 30 天trigger_on_tp: true # 止盈触发冷却trigger_on_sl: false # 止损不触发冷却trigger_on_max_days: true # 超期卖出也触发冷却# 冷却期内是否允许强制买入如手动覆盖allow_override: false# 策略参数strategy:max_positions: 5take_profit_pct: 0.08stop_loss_pct: -0.05max_holding_days: 15initial_capital: 1000000commission_rate: 0.0003stamp_tax_rate: 0.001# 对比实验compare:enabled: truescan_cooling_days: [0, 7, 14, 30, 60, 90] # 扫描不同冷却天数src/data_loader.pydata_loader.py数据加载模块import pandas as pdfrom pathlib import Pathdef load_price_data(filepath: str) - pd.DataFrame:加载日频价格数据预期格式:date,code,open,high,low,close,volume2022-01-03,000001,12.34,12.56,12.10,12.45,1234567df pd.read_csv(filepath, parse_dates[date])df[code] df[code].astype(str).str.zfill(6)return df.set_index([date, code]).sort_index()def get_close_matrix(price_data: pd.DataFrame) - pd.DataFrame:收盘价矩阵: indexdate, columnscodereturn price_data[close].unstack()def generate_mock_prices(n_stocks: int 30,start: str 2022-01-01,end: str 2024-12-31,seed: int 42) - pd.DataFrame:生成模拟价格数据含多次波段便于测试冷却期import numpy as npnp.random.seed(seed)dates pd.date_range(start, end, freqB)codes [f{i:06d} for i in range(n_stocks)]records []for code in codes:# 生成多段趋势模拟一波一波的走势n len(dates)drift np.random.normal(0.0003, 0.015, n)# 叠加波段特征让部分股票有明显的涨一波 → 回调 → 再涨结构if np.random.random() 0.4:# 在随机位置叠加一个脉冲pulse_pos np.random.randint(n // 3, 2 * n // 3)pulse_width np.random.randint(10, 30)for i in range(max(0, pulse_pos - pulse_width), min(n, pulse_pos pulse_width)):drift[i] 0.003close 10 * np.cumprod(1 drift)close np.clip(close, 1.0, None)for d, c in zip(dates, close):records.append({date: d, code: code,open: round(c * 0.998, 2),high: round(c * 1.01, 2),low: round(c * 0.99, 2),close: round(c, 2),volume: int(np.random.exponential(500000))})return pd.DataFrame(records)src/signal_generator.pysignal_generator.pyAI 选股信号生成简化版import pandas as pdimport numpy as npdef generate_daily_signals(close: pd.DataFrame,date: pd.Timestamp,lookback: int 20) - pd.Series:生成每日 AI 信号分0~1简化逻辑实盘替换为真实模型:- 20 日动量- 均线方向- 随机扰动模拟模型不确定性import hashlibscores pd.Series(dtypefloat, indexclose.columns)# 用日期做 seed确定性随机便于回测复现seed_int int(date.strftime(%Y%m%d))np.random.seed(seed_int % (2**32 - 1))for code in close.columns:if code not in close.columns:continue# 动量因子if date in close.index:past_date date - pd.Timedelta(dayslookback)if past_date in close.index:mom (close.loc[date, code] / close.loc[past_date, code]) - 1else:mom 0else:mom 0# 简化为一个 0~1 的分数score 0.4 0.3 * np.clip(mom * 10, -1, 1) 0.3 * np.random.uniform(-0.3, 0.3)score np.clip(score, 0.0, 1.0)scores[code] scorereturn scoresdef get_buy_candidates(signals: pd.Series,threshold: float 0.55,top_n: int 20) - pd.Series:选取高分候选return signals[signals threshold].nlargest(top_n)src/cooldown_manager.py★ 核心模块cooldown_manager.py★ 止盈后冷却期管理器核心功能:1. 记录每只股票的退出原因和时间2. 止盈退出 → 进入冷却期N 天不可再买入3. 止损退出 → 不冷却允许立即再买入4. 冷却期结束后自动恢复为可交易状态5. 提供详细的冷却状态查询和统计import pandas as pdimport numpy as npfrom datetime import timedeltafrom typing import Dict, List, Optional, Tuplefrom enum import Enumimport logginglogging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s)logger logging.getLogger(__name__)class ExitReason(Enum):退出原因枚举TAKE_PROFIT tp # 止盈STOP_LOSS sl # 止损MAX_DAYS max_days # 超期MANUAL manual # 手动FINAL_LIQUIDATE final # 回测结束强制平仓class CooldownStatus(Enum):冷却状态枚举READY ready # 可交易COOLING cooling # 冷却中HOLDING holding # 持仓中class CooldownManager:★ 止盈后冷却期管理器状态字典结构:{000001: {status: CooldownStatus,cooling_until: pd.Timestamp, # 冷却结束日last_exit_date: pd.Timestamp, # 上次退出日last_exit_reason: ExitReason, # 退出原因last_exit_price: float, # 退出价格total_cooling_events: int, # 累计冷却次数total_tp_events: int, # 累计止盈次数}}def __init__(self,cooling_days: int 30,trigger_on_tp: bool True,trigger_on_sl: bool False,trigger_on_max_days: bool True,allow_override: bool False):参数:cooling_days: 冷却期天数默认 30 天trigger_on_tp: 止盈是否触发冷却trigger_on_sl: 止损是否触发冷却通常 Falsetrigger_on_max_days: 超期卖出是否触发冷却allow_override: 是否允许手动覆盖冷却期self.cooling_days cooling_daysself.trigger_on_tp trigger_on_tpself.trigger_on_sl trigger_on_slself.trigger_on_max_days trigger_on_max_daysself.allow_override allow_override# 核心状态存储self.states: Dict[str, Dict] {}# 全局统计self.total_cooling_events 0self.total_blocks 0 # 被冷却拦截的次数logger.info(f冷却期管理器初始化: {cooling_days} 天)logger.info(f 止盈触发: {trigger_on_tp}, 止损触发: {trigger_on_sl})def on_buy(self, code: str, date: pd.Timestamp, price: float):★ 买入事件回调当策略买入一只股票时调用更新状态为 HOLDINGif code not in self.states:self.states[code] self._init_state()state self.states[code]# 检查是否在冷却期if state[status] CooldownStatus.COOLING.value:if not self.allow_override:logger.warning(f[{date.strftime(%Y-%m-%d)}] {code} 仍在冷却期f距离结束还有 {self.days_left(code, date)} 天买入被拦截)return Falseelse:logger.info(f[{date.strftime(%Y-%m-%d)}] {code} 冷却期被手动覆盖)state[status] CooldownStatus.HOLDING.valuestate[last_buy_date] datestate[last_buy_price] pricereturn Truedef on_sell(self,code: str,date: pd.Timestamp,price: float,reason: str):★ 卖出事件回调核心逻辑参数:code: 股票代码date: 卖出日期price: 卖出价格reason: 卖出原因tp / sl / max_days / manual / finalif code not in self.states:self.states[code] self._init_state()state self.states[code]state[status] CooldownStatus.COOLING.valuestate[last_exit_date] datestate[last_exit_price] pricestate[last_exit_reason] reasonstate[total_cooling_events] 1# ★ 核心逻辑根据退出原因决定是否启动冷却期should_cool self._should_trigger_cooling(reason)if should_cool:state[cooling_until] date timedelta(daysself.cooling_days)self.total_cooling_events 1logger.debug(f[{date.strftime(%Y-%m-%d)}] {code} 止盈后进入 {self.cooling_days} 天冷却期 f(至 {state[cooling_until].strftime(%Y-%m-%d)}))else:# 不冷却立即恢复为 readystate[status] CooldownStatus.READY.valuestate[cooling_until] datelogger.debug(f[{date.strftime(%Y-%m-%d)}] {code} {reason} 退出不进入冷却期)if reason tp:state[total_tp_events] 1def is_cooling(self, code: str, date: pd.Timestamp) - bool:★ 核心查询判断某标的当前是否处于冷却期返回:True → 处于冷却期不应买入False → 可买入if code not in self.states:return Falsestate self.states[code]# 如果状态是 HOLDING说明还在持仓不应该再买if state[status] CooldownStatus.HOLDING.value:return True # 持仓中不能重复买入# 如果状态是 COOLING检查冷却期是否已过if state[status] CooldownStatus.COOLING.value:if date state[cooling_until]:# 冷却期结束自动转为 READYstate[status] CooldownStatus.READY.valuereturn Falsereturn True # 仍在冷却期return False # READY 状态def days_left(self, code: str, date: pd.Timestamp) - int:返回某标的冷却期剩余天数0 已结束if code not in self.states:return 0state self.states[code]if state[status] ! CooldownStatus.COOLING.value:return 0delta (state[cooling_until] - date).daysreturn max(0, delta)def filter_candidates(self,candidates: List[str],date: pd.Timestamp) - Tuple[List[str], Dict[str, int]]:★ 核心方法过滤候选列表剔除冷却期标的参数:candidates: AI 模型选出的候选股票列表date: 当前日期返回:(filtered_list, rejection_stats)rejection_stats:{cooling: 被冷却拦截的数量}filtered []rejections {cooling: 0}for code in candidates:if self.is_cooling(code, date):rejections[cooling] 1self.total_blocks 1else:filtered.append(code)return filtered, rejectionsdef _should_trigger_cooling(self, reason: str) - bool:根据退出原因判断是否触发冷却期if reason tp and self.trigger_on_tp:return Trueif reason sl and self.trigger_on_sl:return Trueif reason max_days and self.trigger_on_max_days:return Trueif reason manual:return False # 手动卖出不冷却if reason final:return False # 回测结束强制平仓不冷却return Falsedef _init_state(self) - Dict:初始化标的状态return {status: CooldownStatus.READY.value,cooling_until: None,last_exit_date: None,last_exit_price: None,last_exit_reason: None,last_buy_date: None,last_buy_price: None,total_cooling_events: 0,total_tp_events: 0,}def get_statistics(self) - Dict:返回冷却期统计cooling_count sum(1 for s in self.states.values()if s[status] CooldownStatus.COOLING.value)return {total_cooling_events: self.total_cooling_events,total_blocks: self.total_blocks,currently_cooling: cooling_count,total_tracked: len(self.states),}def print_statistics(self):打印冷却期统计报告stats self.get_statistics()print(f\n{*60})print(f 冷却期统计报告)print(f{*60})print(f 跟踪标的总数: {stats[total_tracked]})print(f 累计冷却事件: {stats[total_cooling_events]})print(f 累计拦截买入: {stats[total_blocks]})print(f 当前仍在冷却: {stats[currently_cooling]})# 打印每只标的的冷却状态if len(self.states) 0:print(f\n 标的冷却状态明细前 15 只:)print(f {代码:8} {状态:10} {退出原因:10} {冷却剩余天:12} {止盈次数:8})print(f {─*60})for i, (code, state) in enumerate(self.states.items()):if i 15:print(f ... 共 {len(self.states)} 只标的)breakstatus_cn {ready: 可交易,cooling: 冷却中,holding: 持仓中}.get(state[status], state[status])days self.days_left(code, pd.Timestamp.now())reason state[last_exit_reason] or --tp_count state[total_tp_events]print(f {code:8} {status_cn:10} {reason:10} {days:12} {tp_count:8})print(f{*60}\n)src/strategy_engine.pystrategy_engine.py策略引擎集成冷却期管理import pandas as pdimport numpy as npfrom src.cooldown_manager import CooldownManager, ExitReasonfrom typing import Dict, Listclass CoolingAwareStrategy:★ 集成冷却期的趋势策略引擎执行顺序1. AI 模型生成候选列表2. ★ 冷却期过滤剔除止盈后 30 天内的标的3. 执行买入4. 持仓期间止盈/止损5. 卖出时通知冷却期管理器def __init__(self,cooldown_manager: CooldownManager,max_positions: int 5,take_profit_pct: float 0.08,stop_loss_pct: float -0.05,max_holding_days: int 15,initial_capital: float 1_000_000,commission_rate: float 0.0003,stamp_tax_rate: float 0.001):self.cd cooldown_managerself.max_pos max_positionsself.tp take_profit_pctself.sl stop_loss_pctself.max_hold max_holding_daysself.comm commission_rateself.tax stamp_tax_rateself.capital initial_capitalself.positions: Dict[str, Dict] {}self.daily_nav: Dict[pd.Timestamp, float] {}self.trade_log: List[Dict] []# 每日被冷却拦截的次数用于统计self.daily_cooling_blocks: Dict[pd.Timestamp, int] {}print(f\n{*60})print(f 冷却期感知策略引擎初始化)print(f 冷却期: {cooldown_manager.cooling_days} 天)print(f 止盈触发冷却: {cooldown_manager.trigger_on_tp})print(f 止损触发冷却: {cooldown_manager.trigger_on_sl})print(f{*60}\n)def run_daily(self,date: pd.Timestamp,close: pd.Series,signals: pd.Series):每日策略执行# 1. 检查持仓止盈止损 to_close []for code, pos in self.positions.items():if code not in close or pd.isna(close[code]) or close[code] 0:continuepnl_pct (close[code] - pos[open_price]) / pos[open_price]days_held (date - pos[open_date]).daysif pnl_pct self.tp:to_close.append((code, tp))elif pnl_pct self.sl:to_close.append((code, sl))elif days_held self.max_hold:to_close.append((code, max_days))for code, reason in to_close:px close.get(code, self.positions[code][open_price])self._close_position(code, date, px, reason)# 2. 生成买入候选 candidates signals[signals 0.55].sort_values(ascendingFalse)# 3. ★ 冷却期过滤 candidate_codes candidates.index.tolist()filtered_codes, rejections self.cd.filter_candidates(candidate_codes, date)self.daily_cooling_blocks[date] rejections.get(cooling, 0)if rejections.get(cooling, 0) 0:logger logging.getLogger(__name__)logger.debug(f[{date.strftime(%Y-%m-%d)}] 冷却期拦截 {rejections[cooling]} 只标的)# 4. 执行买入 if len(self.positions) self.max_pos:for code in filtered_codes:if len(self.positions) self.max_pos:breakif code in self.positions:continueif code not in close or pd.isna(close[code]) or close[code] 0:continue本文代码仅供学习与技术交流不构成任何投资建议股市有风险入市需谨慎利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
