多周期与多币种策略的工程结构

为什么多周期策略特别容易写错

单周期策略的数据只有一张表，时间索引唯一，出错空间有限。一旦引入第二个周期，你手上就有了两条不同频率的时间轴，而这两条轴的对齐方式有一个极易踩错的细节：在15分钟这一格上，当前的4小时K线通常还没有走完。

假设现在是14:15，4小时K线的区间是12:00到16:00。这根K线的收盘价、最高价、成交量此刻都还在变动。如果你直接取这根K线算MA或ATR，你用到的是未来一个多小时才会确定的信息。回测里它会安静地提高胜率，实盘里它根本拿不到。

多币种带来的是另一类错误：状态串台。很多人写循环时把持仓、开仓价、止损价定义成外层变量，循环到第二个币种时覆盖了第一个币种的状态。曲线看起来仍然平滑，但持仓记录已经完全不可信。

时间戳对齐：只用已收盘的大周期数据

对齐的目标是：对任意一个小周期时刻 t，取到的大周期指标必须来自 t 之前已经完全收盘的那根K线。实现上有两种等价写法，推荐第二种，因为它不依赖具体库的默认行为。

第一种是重采样后整体前移一格，也就是先按大周期聚合，再对指标序列做 shift(1)，最后前向填充到小周期索引上。第二种是显式计算每个小周期时刻所属的大周期起点，再手动取上一个起点的值。第二种写法冗长，但每一步都能单独打印检查。

import pandas as pd

def align_higher_tf(df_low: pd.DataFrame,
                    rule: str = "4H") -> pd.DataFrame:
    """把大周期指标安全地对齐到小周期索引上。
    df_low: 索引为 UTC DatetimeIndex 的小周期 OHLCV
    返回: 与 df_low 同索引的大周期指标列
    """
    # 1) 聚合成大周期。label/closed 都用 left，
    #    保证 12:00 这根代表 [12:00, 16:00)
    agg = {"open": "first", "high": "max",
           "low": "min", "close": "last", "volume": "sum"}
    htf = df_low.resample(rule, label="left", closed="left").agg(agg)

    # 2) 在大周期上算指标
    htf["ma"] = htf["close"].rolling(50).mean()
    htf["trend_up"] = htf["close"] > htf["ma"]

    # 3) 关键一步：整体后移一根。
    #    12:00-16:00 这根的结果，最早只能在 16:00 之后使用
    htf_shift = htf[["ma", "trend_up"]].shift(1)

    # 4) 前向填充回小周期索引
    out = htf_shift.reindex(df_low.index, method="ffill")
    out.columns = ["htf_ma", "htf_trend_up"]
    return out


def assert_no_lookahead(df_low, out, rule="4H"):
    """自检：任一时刻的 htf 值，其来源K线必须已收盘。"""
    bucket = df_low.index.floor(rule)
    # 当前所属桶的数据绝不能出现在 out 里
    assert out["htf_ma"].groupby(bucket).nunique().max() <= 1
    return True

python

信号合并：把共振写成显式布尔表达式

多周期共振的常见需求是：大周期确认方向，小周期负责择时。写法上不要用嵌套 if 一层层套，而是把每个条件命名成一个布尔列，最后用一行逻辑运算合成。这样做有三个实际好处。

一是可统计。你能分别算出每个条件单独触发的次数、以及去掉某个条件后收益怎么变，这是判断条件是否真的有贡献的唯一办法。二是可调试。任何一根K线上你都能打印出四个布尔值，立刻知道信号为什么没触发。三是可扩展，加一个周期只是多一列。

另外要给共振设一个时效窗口。大周期在16:00确认转多，小周期在三天后才给出入场信号，这时大周期的确认早已过期。实践中给一个 N 根K线的有效期，超过就作废。

def build_signal(df, htf, valid_bars: int = 12):
    d = df.join(htf)

    # --- 每个条件一列，命名清楚 ---
    d["c_htf_trend"] = d["htf_trend_up"].fillna(False)
    d["c_ltf_cross"] = (d["close"] > d["ma20"]) & \
                       (d["close"].shift(1) <= d["ma20"].shift(1))
    d["c_vol"] = d["volume"] > d["volume"].rolling(96).mean()
    d["c_not_extreme"] = d["rsi14"] < 75

    # --- 大周期确认的时效性 ---
    #  确认后 valid_bars 根内有效
    fresh = d["c_htf_trend"].rolling(valid_bars).max().astype(bool)

    d["long_entry"] = (fresh
                       & d["c_ltf_cross"]
                       & d["c_vol"]
                       & d["c_not_extreme"])
    return d


def condition_contribution(d, cols, target="long_entry"):
    """逐个去掉条件，看信号数量变化，判断哪个条件在真正筛选"""
    base = int(d[target].sum())
    rep = {}
    for c in cols:
        others = [x for x in cols if x != c]
        n = int(d[others].all(axis=1).sum())
        rep[c] = {"base": base, "without": n}
    return rep

python

状态管理：每个币种一个独立状态机

多币种回测的正确抽象是：一个 Position 对象负责单个 symbol 的全部状态，一个 Portfolio 负责资金与总风险约束。信号计算是无状态的纯函数，状态只在执行层改变。这条边界划清了，串台问题基本不会发生。

单个币种需要维护的状态比新手预期的多：是否持仓、方向、开仓价、开仓时间、当前止损位（移动止损会变）、已加仓次数、本轮最高浮盈（用于回撤止盈）。这些都是 per-symbol 的，任何一个漏掉都会导致逻辑退化。

组合层要管的是跨币种约束：同时最多持仓几个、总名义敞口上限、相关性过高的币种不同时开。加密市场里绝大多数山寨币与 BTC 高度同向，所谓「分散到十个币」在下跌时往往是同一笔风险。

from dataclasses import dataclass, field
from typing import Dict, Optional

@dataclass
class Position:
    symbol: str
    side: int = 0            # 0 空仓, 1 多, -1 空
    qty: float = 0.0
    entry_px: float = 0.0
    entry_ts: Optional[object] = None
    stop_px: float = 0.0
    peak_pnl: float = 0.0    # 本轮最高浮盈, 用于回撤止盈
    adds: int = 0

    @property
    def is_open(self) -> bool:
        return self.side != 0

    def reset(self):
        self.side, self.qty, self.adds = 0, 0.0, 0
        self.entry_px = self.stop_px = self.peak_pnl = 0.0
        self.entry_ts = None


@dataclass
class Portfolio:
    cash: float
    max_positions: int = 5
    risk_per_trade: float = 0.01      # 单笔风险 1%
    positions: Dict[str, Position] = field(default_factory=dict)

    def get(self, symbol: str) -> Position:
        # 关键: 按 symbol 惰性创建, 绝不共用
        if symbol not in self.positions:
            self.positions[symbol] = Position(symbol)
        return self.positions[symbol]

    def open_count(self) -> int:
        return sum(p.is_open for p in self.positions.values())

    def can_open(self) -> bool:
        return self.open_count() < self.max_positions

    def size_by_risk(self, equity, entry, stop) -> float:
        risk_amt = equity * self.risk_per_trade
        per_unit = abs(entry - stop)
        return 0.0 if per_unit <= 0 else risk_amt / per_unit


# 主循环: 时间在外, 币种在内, 状态从 portfolio 取
def run(bars_by_symbol, pf: Portfolio, timeline):
    for ts in timeline:
        for sym, df in bars_by_symbol.items():
            if ts not in df.index:
                continue          # 该币此刻无数据, 跳过而非补零
            row = df.loc[ts]
            pos = pf.get(sym)     # 每个币独立状态
            step(pos, pf, row, ts)

python

工程结构：四层分工与目录组织

代码规模超过几百行之后，分层比技巧重要。推荐的划分是数据层、指标层、信号层、执行层，规则是上层可以调用下层，下层绝不反向依赖。这样任何一层都能被单独测试。

目录上对应四个包，再加 configs/ 存参数、tests/ 存单元测试、reports/ 存回测产出。参数一律走配置文件，不要散落在代码里的魔法数字——做参数扫描时你会感谢自己，具体方法见过拟合识别。

撮合与绩效部分的具体实现，可以参考从零搭建最小回测引擎；那篇里的撮合循环可以直接套在本文的 Portfolio 之上。

多币种数据的三个坑

第一个坑是上线时间不同。ETH 有多年数据，某个新币只有三个月。如果你直接取时间交集，等于只用最短那个币的历史；如果直接取并集又会在早期出现大量 NaN。正确做法是各币种独立跑自己的可用区间，在组合层按实际在场币种数计算权重。

第二个坑是退市与改名。回测样本里只包含今天还在交易的币种，就是典型的幸存者偏差，结果会系统性偏乐观。至少要在报告里注明样本构成，最好保留已下架币种的历史数据。

第三个坑是不同币种的最小下单量和价格精度不同。回测里用浮点数随便下 0.0037 个 BTC 没问题，实盘会被拒单。建议在执行层加一次量化取整，并把因取整而无法下单的信号记录下来。

第三个坑对小资金账户尤其明显。若账户只有几千 USDT 又要分散到多个币种，单笔名义金额可能低于交易所的最小下单金额，信号会被静默丢弃。把这类信号计入统计后，你才知道这套配置实际能执行的信号只有原设计的多少比例。

怎么验证你的实现是对的

工程正确性不能靠肉眼看曲线，要靠可重复的检查。下面五项是最低要求，每次改动核心逻辑后都应该跑一遍，成本很低但能拦住绝大多数隐蔽错误。

1. 单币种退化测试：把多币种引擎只喂一个币，结果必须和单币版本逐笔一致。
2. 时间倒序测试：故意把大周期 shift 去掉，收益应显著上升。如果没变化，说明大周期条件其实没起作用。
3. 状态泄漏测试：跑两个完全独立的币，把其中一个的行情整体乘以 2，另一个的成交记录必须一字不变。
4. 时间戳边界测试：单独打印 15:45、16:00、16:15 三个时刻取到的 htf 值，确认在 16:00 之后才切换。
5. 成交明细抽查：随机抽 10 笔成交，手工核对开仓时刻的指标值是否满足入场条件。

这五项做完，剩下的分歧就是策略逻辑层面的了，可以进入Walk-Forward 验证阶段。在此之前的任何收益数字都不值得讨论，因为你还不确定它们是策略的功劳还是bug的功劳。

最后提醒一句关于工程成本的现实：多周期多币种的代码量通常是单周期单币的三到五倍，其中大半用在数据与状态管理上，而不是策略思路。这部分投入无法省略，也无法靠更聪明的信号绕过。