Sequoia-X 新增策略：四步把私有逻辑接进流水线

1. 新建策略文件

   在 sequoia_x/strategy/ 下新增 my_signal.py，导入 BaseStrategy、get_logger 与 pandas。文件名与类名风格参照现有策略（如 ma_volume.py 对应 MaVolumeStrategy）。

2. 实现 run() 并返回代码列表

   用 self.engine.get_local_symbols() 拿全市场代码，逐只 self.engine.get_ohlcv(symbol) 取日 K，做向量化计算后把命中的代码 append 进列表返回。需要 60 日均线这类长窗口时，先判断 len(df) 是否够。

3. 设置 webhook_key

   给类加一行 webhook_key = "my_signal"，并在 .env 里补 STRATEGY_WEBHOOK_MY_SIGNAL（不配则回落到默认 Webhook，也能跑）。

4. 注册到 main.py

   在导入区加一行 import，在 strategies 列表里追加实例 YourSignalStrategy(engine=engine, settings=settings)。列表顺序就是执行顺序。

5. 跑一次验证

   先跑 python main.py 看日志里是否出现「YourSignalStrategy 选出 N 只股票」，再确认命中的推送落在了预期群里。空命中时只会有日志、不会有卡片。

import sqlite3, pandas as pd

HOLD_DAYS = [1, 5, 10, 20]                  # 持有周期（交易日）
FEE, TAX, SLIP = 0.00025, 0.0005, 0.001     # 佣金 / 卖出印花税 / 单边滑点（按你的券商改）

conn = sqlite3.connect("data/sequoia_v2.db")
bars = pd.read_sql(
    "SELECT symbol, date, open, high, low, close, volume, turnover "
    "FROM stock_daily ORDER BY symbol, date", conn)

def find_signals(bars):
#   把你的策略条件写在这里；硬约束：只能用当日及之前的数据
    out = []
    for symbol, g in bars.groupby("symbol"):
        if len(g) < 21:
            continue
        g = g.copy()
        g["high_20"] = g["high"].shift(1).rolling(20).max()   # shift(1) 防未来函数
        for i in range(20, len(g) - 1):                       # 留出次日，保证能成交
            row, prev = g.iloc[i], g.iloc[i - 1]
            if (row["close"] > row["high_20"] and row["turnover"] > 1e8
                    and row["close"] > row["open"] and row["close"] > prev["close"]):
                out.append((symbol, i))                        # (股票, 信号在序列中的位置)
    return out, bars

def forward_return(g, i, n):
    # 次日开盘买入 → 第 n 个交易日开盘卖出（若样本不够则丢弃）
    if i + n >= len(g):
        return None
    buy, sell = g["open"].iloc[i + 1], g["open"].iloc[i + n]
    gross = sell / buy - 1
    cost = FEE * 2 + TAX + SLIP * 2          # 双边佣金 + 卖出印花税 + 双边滑点
    return gross - cost

signals, bars = find_signals(bars)
rows = []
for symbol, i in signals:
    g = bars[bars["symbol"] == symbol]
    for n in HOLD_DAYS:
        r = forward_return(g, i, n)
        if r is not None:
            rows.append({"symbol": symbol, "signal_idx": i, "hold": n, "ret": r})

res = pd.DataFrame(rows)
print(res.groupby("hold")["ret"].agg(["count", "mean", "median",
      lambda s: (s > 0).mean()]).rename(columns={"<lambda_0>": "win_rate"}))

1. 先把区间切成两段，再动手改参数

   # 例：库内 2024-01-01 起，按时间切成 70% / 30%
   cut = "2025-06-30"
   train = bars[bars["date"] <= cut]     # 样本内：可以反复试参数
   test  = bars[bars["date"] >  cut]     # 样本外：只能跑一次
   # 纪律：参数只在 train 上挑，选定后到 test 上只跑一次并如实记录

   最容易违反的一点是「在样本外跑完不满意又回去调参」——那等于把样本外也用成了样本内。要改就重新切一段新的样本外区间。

2. 记录每一次参数改动，连同改动的理由

   # 建议在策略类里保留一份改动记录，而不是只留最终值
   class MyStrategy(BaseStrategy):
       webhook_key = "my_signal"
       window = 20          # 2026-09-20 由 15 调整为 20：15 在样本内命中过多，改用月线级别
       threshold = 1.5      # 保持默认，未调

   没有理由的调参最后只剩一堆数字。半年后回看，你无法判断某个阈值是「有依据」还是「试出来的」。

3. 用样本外结果说话，并公布失败的那部分

   print(res.groupby("hold")["ret"].agg(["count", "mean", "median"]))
   # 同时输出：去重后的样本量、无法成交样本占比、分年度收益
   # 结论里必须包含「样本外表现」与「未通过的假设」

   本站对任何策略都不给有效性结论，原因就在这里：形态类规则的有效性会随市场环境变化，样本外表现才是决定它能不能用的依据。所有结果仅供研究与学习，不构成投资建议。