聚焦“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 / holding
'cooling_until': Timestamp, # 冷却结束日期
'last_exit_date': Timestamp, # 上次退出日期
'last_exit_reason': 'tp', # tp / sl / manual
'days_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.txt"
pandas>=1.5
numpy>=1.21
matplotlib>=3.5
seaborn>=0.12
scipy>=1.9
pyyaml>=6.0
"config.yaml"
# 止盈冷却期配置
# ★ 冷却期参数
cooldown:
enabled: true
cooling_days: 30 # 止盈后冷却 30 天
trigger_on_tp: true # 止盈触发冷却
trigger_on_sl: false # 止损不触发冷却
trigger_on_max_days: true # 超期卖出也触发冷却
# 冷却期内是否允许"强制买入"（如手动覆盖）
allow_override: false
# 策略参数
strategy:
max_positions: 5
take_profit_pct: 0.08
stop_loss_pct: -0.05
max_holding_days: 15
initial_capital: 1000000
commission_rate: 0.0003
stamp_tax_rate: 0.001
# 对比实验
compare:
enabled: true
scan_cooling_days: [0, 7, 14, 30, 60, 90] # 扫描不同冷却天数
"src/data_loader.py"
"""
data_loader.py
数据加载模块
"""
import pandas as pd
from pathlib import Path
def load_price_data(filepath: str) -> pd.DataFrame:
"""
加载日频价格数据
预期格式:
date,code,open,high,low,close,volume
2022-01-03,000001,12.34,12.56,12.10,12.45,1234567
"""
df = 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:
"""收盘价矩阵: index=date, columns=code"""
return 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 np
np.random.seed(seed)
dates = pd.date_range(start, end, freq='B')
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.003
close = 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.py"
"""
signal_generator.py
AI 选股信号生成（简化版）
"""
import pandas as pd
import numpy as np
def generate_daily_signals(
close: pd.DataFrame,
date: pd.Timestamp,
lookback: int = 20
) -> pd.Series:
"""
生成每日 AI 信号分（0~1）
简化逻辑（实盘替换为真实模型）:
- 20 日动量
- 均线方向
- 随机扰动（模拟模型不确定性）
"""
import hashlib
scores = pd.Series(dtype=float, index=close.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(days=lookback)
if past_date in close.index:
mom = (close.loc[date, code] / close.loc[past_date, code]) - 1
else:
mom = 0
else:
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] = score
return scores
def 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 pd
import numpy as np
from datetime import timedelta
from typing import Dict, List, Optional, Tuple
from enum import Enum
import logging
logging.basicConfig(level=logging.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: 止损是否触发冷却（通常 False）
trigger_on_max_days: 超期卖出是否触发冷却
allow_override: 是否允许手动覆盖冷却期
"""
self.cooling_days = cooling_days
self.trigger_on_tp = trigger_on_tp
self.trigger_on_sl = trigger_on_sl
self.trigger_on_max_days = trigger_on_max_days
self.allow_override = allow_override
# 核心状态存储
self.states: Dict[str, Dict] = {}
# 全局统计
self.total_cooling_events = 0
self.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):
"""
★ 买入事件回调
当策略买入一只股票时调用，更新状态为 HOLDING
"""
if 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 False
else:
logger.info(f"[{date.strftime('%Y-%m-%d')}] {code} 冷却期被手动覆盖")
state['status'] = CooldownStatus.HOLDING.value
state['last_buy_date'] = date
state['last_buy_price'] = price
return True
def on_sell(
self,
code: str,
date: pd.Timestamp,
price: float,
reason: str
):
"""
★ 卖出事件回调（核心逻辑）
参数:
code: 股票代码
date: 卖出日期
price: 卖出价格
reason: 卖出原因（'tp' / 'sl' / 'max_days' / 'manual' / 'final'）
"""
if code not in self.states:
self.states[code] = self._init_state()
state = self.states[code]
state['status'] = CooldownStatus.COOLING.value
state['last_exit_date'] = date
state['last_exit_price'] = price
state['last_exit_reason'] = reason
state['total_cooling_events'] += 1
# ★ 核心逻辑：根据退出原因决定是否启动冷却期
should_cool = self._should_trigger_cooling(reason)
if should_cool:
state['cooling_until'] = date + timedelta(days=self.cooling_days)
self.total_cooling_events += 1
logger.debug(
f"[{date.strftime('%Y-%m-%d')}] {code} 止盈后进入 {self.cooling_days} 天冷却期 "
f"(至 {state['cooling_until'].strftime('%Y-%m-%d')})"
)
else:
# 不冷却：立即恢复为 ready
state['status'] = CooldownStatus.READY.value
state['cooling_until'] = date
logger.debug(
f"[{date.strftime('%Y-%m-%d')}] {code} {reason} 退出，不进入冷却期"
)
if reason == 'tp':
state['total_tp_events'] += 1
def is_cooling(self, code: str, date: pd.Timestamp) -> bool:
"""
★ 核心查询：判断某标的当前是否处于冷却期
返回:
True → 处于冷却期，不应买入
False → 可买入
"""
if code not in self.states:
return False
state = self.states[code]
# 如果状态是 HOLDING，说明还在持仓（不应该再买）
if state['status'] == CooldownStatus.HOLDING.value:
return True # 持仓中，不能重复买入
# 如果状态是 COOLING，检查冷却期是否已过
if state['status'] == CooldownStatus.COOLING.value:
if date >= state['cooling_until']:
# 冷却期结束，自动转为 READY
state['status'] = CooldownStatus.READY.value
return False
return True # 仍在冷却期
return False # READY 状态
def days_left(self, code: str, date: pd.Timestamp) -> int:
"""返回某标的冷却期剩余天数（0 = 已结束）"""
if code not in self.states:
return 0
state = self.states[code]
if state['status'] != CooldownStatus.COOLING.value:
return 0
delta = (state['cooling_until'] - date).days
return 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'] += 1
self.total_blocks += 1
else:
filtered.append(code)
return filtered, rejections
def _should_trigger_cooling(self, reason: str) -> bool:
"""根据退出原因判断是否触发冷却期"""
if reason == 'tp' and self.trigger_on_tp:
return True
if reason == 'sl' and self.trigger_on_sl:
return True
if reason == 'max_days' and self.trigger_on_max_days:
return True
if reason == 'manual':
return False # 手动卖出不冷却
if reason == 'final':
return False # 回测结束强制平仓不冷却
return False
def _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)} 只标的")
break
status_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.py"
"""
strategy_engine.py
策略引擎：集成冷却期管理
"""
import pandas as pd
import numpy as np
from src.cooldown_manager import CooldownManager, ExitReason
from typing import Dict, List
class 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_manager
self.max_pos = max_positions
self.tp = take_profit_pct
self.sl = stop_loss_pct
self.max_hold = max_holding_days
self.comm = commission_rate
self.tax = stamp_tax_rate
self.capital = initial_capital
self.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:
continue
pnl_pct = (close[code] - pos['open_price']) / pos['open_price']
days_held = (date - pos['open_date']).days
if 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(ascending=False)
# === 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:
break
if code in self.positions:
continue
if code not in close or pd.isna(close[code]) or close[code] <= 0:
continue
本文代码仅供学习与技术交流，不构成任何投资建议，股市有风险，入市需谨慎！
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
