文章

凌晨 3 点，你盯着屏幕上的回测结果，感到一阵寒意。

策略逻辑没问题，代码跑通了，收益曲线却像一条被揉皱的正弦波——亏损的那几周，恰好对应着港股上涨、A股横盘的时间段。

问题不在策略本身。你的回测假设三个市场在同一套时间轴上运行，但它们从来就不是。港股在国庆节休市，A股在春节停摆，美股的"今天"和中国的"今天"差了整整 13 个小时。更要命的是，当你想把美股财报数据和港股衍生品数据按时间对齐时，发现它们的"同一时刻"根本不存在——美股收盘时，港股已经收了两个小时。

这不是时区问题，这是一个多市场回测的坐标系问题。

本文拆解三个核心障碍：UTC 统一、交易日历对齐、Pandas merge 策略，并给出生产级的多市场数据对齐代码。你会看到，不是把三个市场的数据塞进一张表就叫"对齐"，真正的对齐发生在时区层、交易日层和数据密度层三个维度同时收敛之后。

---

一、问题拆解：为什么多市场回测比单市场复杂三个量级

单市场回测的数据对齐是平凡问题：同一个时区、同一个交易日定义、同一种交易状态（开/闭）。多市场回测则面临三层叠加的复杂性。

1.1 时区层：UTC 不是时间的终点，而是起点

时区转换有两个方向，每个方向都有坑。

第一个方向：本地时间 → UTC。美股数据通常以 EST/EDT 存储，港股以 HKT，A 股以 CST。当这些数据进入同一个 Pandas DataFrame 时，如果不做 UTC 统一，你以为自己在做"跨日分析"，实际上在做"跨时区幻觉"。

举例：美股收盘时间 16:00 EST 对应 UTC 次日凌晨 4:00，港股收盘时间 16:00 HKT 对应 UTC 同日 8:00。这两个"16:00"在本地时间上看只差一天，在 UTC 时间轴上却差了 20 个小时。如果你的回测引擎用本地时间做日线级别 join，会把美股收盘数据错误地对齐到港股次日的数据槽位上——策略信号在 UTC 时间轴上错位了 20 小时，等港股下一根 K 线"看到"信号时，价格早已走远。

第二个方向：UTC → 本地时间。当回测引擎输出结果时，你需要把 UTC 时间转换回各市场的本地时间用于展示和风控。如果这一步处理不一致，同一个 UTC 时刻会对应不同的"本地交易日"，导致风控系统误判持仓日期。

1.2 交易日历层：三个市场，没有一天是完全重叠的

全球没有一个"全球交易日"的概念。每个市场的交易日定义不仅取决于日期，还取决于：

• 各自的节假日日历：美股休市日（感恩节、圣诞节）、港股休市日（国庆节假期，但与大陆不完全一致）、A 股休市日（春节假期长度每年不同）。更复杂的是，同一天的"交易日"在不同市场间可能是不同含义——2 月 14 日对美股是交易日，对 A 股可能是春节假期中的某一天。
• 夏令时切换：美股每年 3 月第二个周日到 11 月第一个周日之间切换 EDT，其余时间切换 EST。A 股和港股没有夏令时。一年两次切换发生时，UTC 偏移量会突变，导致时间戳在切换前后的"同一 UTC 时刻"对应完全不同的本地时间。
• 日内交易日定义：A 股日内分上午盘（9:30-11:30）和下午盘（13:00-15:00），中间间隔 90 分钟。港股日内连续竞价（9:30-12:00 和 13:00-16:00）。美股日内最复杂：盘前 4:00-9:30，常规交易 9:30-16:00，盘后 16:00-20:00。当你在 UTC 时间轴上做事件对齐时，这些碎片化的时间窗口如果不显式建模，会导致"数据明明在那里，为什么 join 出来是空的"的困惑。

1.3 数据密度层：每个市场的数据分辨率不同

市场	数据类型	TickDB 支持情况	数据密度
美股	历史 K 线	✅ 10 年级别	1 分钟 / 5 分钟 / 日线
港股	历史 K 线	✅	1 分钟起
A 股	历史 K 线	✅	1 分钟起
美股	tick 级逐笔成交	❌ 不支持	-
港股	tick 级逐笔成交	✅	按交易所推送频率

当你在同一个回测中同时使用美股日线和港股 tick 数据时，merge 策略必须考虑不对等的数据密度——是按 tick 数据的时间戳对齐日线数据（取收盘价），还是按日线数据的时间戳切片 tick 数据（截取日内片段）？这两种方式在量化语义上完全不同。

---

二、架构总览：三阶段数据对齐流水线

多市场数据对齐不是一次性的 pd.merge()，而是一条三阶段流水线：

┌─────────────────────────────────────────────────────────┐
│  Stage 1: 时区标准化                                     │
│  所有市场数据 → 统一 UTC 时间戳                           │
│  关键：显式处理夏令时切换，保留 UTC offset 信息            │
└──────────────────────────┬──────────────────────────────┘
                           ▼
┌─────────────────────────────────────────────────────────┐
│  Stage 2: 交易日历对齐                                    │
│  UTC 时间轴 + 交易日历过滤 + 交易时段标记                  │
│  关键：每个市场独立建模日内时段，跨市场时生成虚拟交易日      │
└──────────────────────────┬──────────────────────────────┘
                           ▼
┌─────────────────────────────────────────────────────────┐
│  Stage 3: 数据密度对齐与 Merge                            │
│  统一分辨率 → 时间窗口对齐 → 缺失值处理                     │
│  关键：按策略需求选择对齐策略（逐笔对齐 / 窗口切片 / 重采样） │
└─────────────────────────────────────────────────────────┘

核心设计原则：每个阶段处理一个维度的复杂性，不要试图在一个 Pandas 操作中同时解决时区、交易日和密度三个问题。分阶段的好处是，每一步的输出都可以单独验证、回滚和调试。

---

三、生产级代码：完整的三阶段对齐流水线

以下代码基于 Python 3.10+，依赖 pandas>=2.0、numpy、pytz。代码使用 TickDB REST API 获取历史 K 线数据，完整实现三阶段对齐流程。

3.1 环境配置与依赖

import os
import time
import json
import random
import requests
import numpy as np
import pandas as pd
from datetime import datetime, timezone, timedelta
from typing import Optional

# ─────────────────────────────────────────────────────────
# TickDB API 配置
# ⚠️ 生产环境建议将 API Key 存储在 .env 文件中，
#    避免硬编码在代码仓库。
# ─────────────────────────────────────────────────────────
API_KEY = os.environ.get("TICKDB_API_KEY")
BASE_URL = "https://api.tickdb.ai/v1"

def tickdb_headers() -> dict:
    """标准 TickDB 请求头"""
    if not API_KEY:
        raise EnvironmentError("请设置环境变量 TICKDB_API_KEY")
    return {"X-API-Key": API_KEY}


def handle_tickdb_error(response: requests.Response) -> dict:
    """TickDB 标准错误处理"""
    try:
        payload = response.json()
    except ValueError:
        payload = {"code": -1, "message": response.text}

    code = payload.get("code", 0)
    if code == 0:
        return payload.get("data") or {}

    error_map = {
        1001: "API Key 无效",
        1002: "API Key 缺失",
        2002: "交易品种不存在（请检查 symbols/available）",
        3001: "请求频率超限",
    }
    msg = error_map.get(code, f"未知错误 {code}")
    if code == 3001:
        retry_after = int(response.headers.get("Retry-After", 5))
        time.sleep(retry_after)
        return None
    raise RuntimeError(f"[TickDB 错误] {code}: {msg}")

3.2 第一阶段：时区标准化

时区标准化的核心逻辑：

1. 所有数据统一转换为 UTC 时间戳（tz_localize → tz_convert）
2. 显式记录每个市场的本地时区，以便在夏令时切换点验证连续性
3. 夏令时切换时插入特殊标记，防止回测引擎在切换瞬间产生错误的信号

import pytz


class TimezoneStandardizer:
    """
    多市场时区标准化器。
    将各市场的本地时间戳统一转换为 UTC，
    并在夏令时切换点插入元数据标记。
    """

    # 各市场的本地时区（考虑夏令时）
    LOCAL_TZ = {
        "US": "America/New_York",   # EST/EDT，自动处理夏令时
        "HK": "Asia/Hong_Kong",     # HKT，无夏令时
        "CN": "Asia/Shanghai",      # CST，无夏令时
    }

    def __init__(self):
        self.tz_cache = {k: pytz.timezone(v) for k, v in self.LOCAL_TZ.items()}

    def normalize(
        self,
        df: pd.DataFrame,
        market_col: str,
        timestamp_col: str,
        source_tz: Optional[str] = None,
    ) -> pd.DataFrame:
        """
        将 DataFrame 中的时间戳统一为 UTC。

        Parameters
        ----------
        df : pd.DataFrame
            包含市场标识和时间戳的 DataFrame
        timestamp_col : str
            时间戳列名（支持 datetime 或字符串格式）
        source_tz : str, optional
            源时区。如果为 None，则根据 market_col 值推断。
        """
        df = df.copy()
        tz_name = source_tz or self.LOCAL_TZ.get(market_col.upper())
        if tz_name is None:
            raise ValueError(f"未知市场代码: {market_col}")

        local_tz = self.tz_cache[market_col.upper()]

        # 转换逻辑：先 localize（将 naive datetime 关联到指定时区），
        # 再 convert to UTC
        ts = df[timestamp_col]
        if isinstance(ts.iloc[0], str):
            ts = pd.to_datetime(ts)

        if ts.dt.tz is None:
            # naive datetime → localize → convert to UTC
            ts_utc = ts.dt.tz_localize(local_tz).dt.tz_convert("UTC")
        else:
            # aware datetime → 直接转换到 UTC
            ts_utc = ts.dt.tz_convert("UTC")

        df[f"{timestamp_col}_utc"] = ts_utc
        df[f"{timestamp_col}_src_tz"] = market_col.upper()

        return df

    def detect_dst_transitions(
        self, df: pd.DataFrame, timestamp_col: str, window_days: int = 7
    ) -> pd.DataFrame:
        """
        检测 UTC 时间戳序列中的夏令时切换点。

        夏令时切换会导致 UTC 偏移量突变，
        对需要精确时间对齐的回测策略（如高频事件驱动）
        必须在切换点重新校准信号时间。

        Returns
        -------
        DataFrame 包含切换点前后的 UTC 时间和偏移量变化。
        """
        ts_utc = pd.to_datetime(df[timestamp_col]).dt.tz_convert("UTC")
        offsets = ts_utc.map(lambda x: x.utcoffset().total_seconds())

        # 计算相邻时间点之间的偏移量变化
        offset_diff = offsets.diff().abs()
        dst_mask = offset_diff > 0

        if dst_mask.sum() == 0:
            return pd.DataFrame(columns=["dst_transition_utc", "offset_before", "offset_after"])

        dst_transitions = ts_utc[dst_mask].reset_index(drop=True)
        transitions_df = pd.DataFrame({
            "dst_transition_utc": dst_transitions,
            "offset_before_sec": offsets.diff()[dst_mask].values,
            "offset_after_sec": offsets.shift(-1)[dst_mask].values,
        })
        return transitions_df

3.3 第二阶段：交易日历对齐

这一阶段需要为每个市场单独构建交易日历，并将 UTC 时间轴与各市场的交易日定义对齐。核心挑战在于：不同市场的节假日定义不同，不能用一个统一的"交易日"过滤条件处理所有市场。

from pandas.tseries.offsets import CustomBusinessDay
from typing import Dict, List


class MultiMarketCalendar:
    """
    多市场交易日历管理器。

    每个市场维护独立的节假日列表，
    支持跨市场"共同交易日"查询。
    """

    def __init__(self):
        # 各市场的基本工作日定义（Mon-Fri）
        self._base_holidays: Dict[str, List[pd.Timestamp]] = {
            "US": self._get_us_holidays(),
            "HK": self._get_hk_holidays(),
            "CN": self._get_cn_holidays(),
        }
        self.calendars: Dict[str, CustomBusinessDay] = {}
        self._build_calendars()

    def _get_us_holidays(self) -> List[pd.Timestamp]:
        """美股主要节假日（NYSE 官方规则简化版）"""
        # ⚠️ 生产环境应使用 pandas_market_calendars 库读取 NYSE 官方日历
        # 此处仅作示例，包含核心不可移休市日
        holidays = []
        for year in range(2015, 2027):
            holidays.extend([
                pd.Timestamp(f"{year}-01-01"),   # 元旦
                pd.Timestamp(f"{year}-07-04"),  # 独立日
                pd.Timestamp(f"{year}-12-25"),  # 圣诞节
                # 感恩节（11月第四个周四）
                pd.Timestamp(year=year, month=11, day=1) + pd.offsets.WeekOfMonth(week=3, weekday=3),
            ])
        return holidays

    def _get_hk_holidays(self) -> List[pd.Timestamp]:
        """港股主要节假日"""
        # ⚠️ 实际应接入 HKEX 官方日历
        holidays = []
        for year in range(2015, 2027):
            holidays.extend([
                pd.Timestamp(f"{year}-01-01"),
                pd.Timestamp(f"{year}-04-05"),  # 清明
                pd.Timestamp(f"{year}-05-01"),
                pd.Timestamp(f"{year}-07-01"),   # 香港回归
                pd.Timestamp(f"{year}-10-01"),   # 国庆（部分假期）
                pd.Timestamp(f"{year}-12-25"),
                pd.Timestamp(f"{year}-12-26"),   # 圣诞节次日
            ])
        return holidays

    def _get_cn_holidays(self) -> List[pd.Timestamp]:
        """A 股主要节假日"""
        # ⚠️ A 股节假日每年不同，此处仅示意结构
        # 生产环境应使用 akshare 或 tushare 获取实时节假日数据
        holidays = []
        for year in range(2015, 2027):
            holidays.extend([
                pd.Timestamp(f"{year}-01-01"),
                pd.Timestamp(f"{year}-05-01"),
                pd.Timestamp(f"{year}-10-01"),   # 国庆（通常 7 天）
            ])
        return holidays

    def _build_calendars(self):
        """为每个市场构建 CustomBusinessDay 对象"""
        for market, holidays in self._base_holidays.items():
            self.calendars[market] = CustomBusinessDay(holidays=holidays)

    def is_trading_day(self, date: pd.Timestamp, market: str) -> bool:
        """判断某日期是否为指定市场的交易日"""
        cal = self.calendars.get(market.upper())
        if cal is None:
            raise ValueError(f"未知市场: {market}")
        try:
            return pd.Timestamp(date, tz="UTC").tz_convert(self.LOCAL_TZ[market]) not in holidays
        except Exception:
            return True

    def get_common_trading_days(
        self,
        start_date: pd.Timestamp,
        end_date: pd.Timestamp,
        markets: List[str],
    ) -> pd.DatetimeIndex:
        """
        获取所有指定市场同时开放的交易日列表。

        用于需要"跨市场同时信号"的策略——
        例如美股财报发布后 30 分钟内同步做空港股相关衍生品。
        如果某天港股休市而美股正常交易，该天不会被纳入共同交易日。
        """
        if len(markets) == 1:
            market = markets[0].upper()
            cal = self.calendars[market]
            return pd.date_range(start=start_date, end=end_date, freq=cal)

        # 多市场：取交集
        common_days = None
        for market in markets:
            market_cal = self.calendars[market.upper()]
            market_days = pd.date_range(start=start_date, end=end_date, freq=market_cal)
            if common_days is None:
                common_days = set(market_days)
            else:
                common_days &= set(market_days)

        return pd.DatetimeIndex(sorted(common_days))

    # 引用 TimezoneStandardizer 的时区定义
    LOCAL_TZ = TimezoneStandardizer.LOCAL_TZ

3.4 第三阶段：数据对齐与 Merge 策略

对齐策略的选择取决于你的回测语义。以下是三种主流场景对应的 Pandas merge 策略：

def align_and_merge(
    us_df: pd.DataFrame,
    hk_df: pd.DataFrame,
    cn_df: pd.DataFrame,
    align_strategy: str = "common_trading_day",
    lookback_days: int = 5,
) -> pd.DataFrame:
    """
    多市场数据对齐主函数。

    Parameters
    ----------
    align_strategy : str
        - "common_trading_day": 按共同交易日对齐（默认，适用日线策略）
        - "utc_timestamp": 按 UTC 时间戳精确对齐（适用高频事件驱动）
        - "resample_1h": 重采样到统一 1H 频率后对齐（适用多市场信号融合）
    lookback_days : int
        对齐时向前搜索的天数（应对跨市场信号延迟）

    Returns
    -------
    对齐后的 DataFrame，UTC 索引，多市场列合并。
    """

    # ── Step 1: 统一 UTC 时间戳 ──────────────────────────
    us_df = TimezoneStandardizer().normalize(us_df, "US", "timestamp")
    hk_df = TimezoneStandardizer().normalize(hk_df, "HK", "timestamp")
    cn_df = TimezoneStandardizer().normalize(cn_df, "CN", "timestamp")

    # ── Step 2: 重命名为统一列名 ────────────────────────
    us_df = us_df.rename(columns={
        "close": "us_close",
        "volume": "us_volume",
        "timestamp_utc": "utc_time",
    })
    hk_df = hk_df.rename(columns={
        "close": "hk_close",
        "volume": "hk_volume",
        "timestamp_utc": "utc_time",
    })
    cn_df = cn_df.rename(columns={
        "close": "cn_close",
        "volume": "cn_volume",
        "timestamp_utc": "utc_time",
    })

    if align_strategy == "common_trading_day":
        # ── 策略 A：按共同交易日对齐 ────────────────────
        # 适用于日线级别的多市场策略（如跨市场动量）
        start = us_df["utc_time"].min()
        end = us_df["utc_time"].max()

        common_days = MultiMarketCalendar().get_common_trading_days(
            start, end, ["US", "HK", "CN"]
        )
        common_days_utc = pd.DatetimeIndex(common_days).tz_localize("UTC")

        # 将每日数据聚合到 UTC 00:00（市场本地收盘时对应的 UTC 时间）
        us_daily = us_df.resample("D", on="utc_time").agg({
            "us_close": "last",
            "us_volume": "sum",
        }).dropna()

        hk_daily = hk_df.resample("D", on="utc_time").agg({
            "hk_close": "last",
            "hk_volume": "sum",
        }).dropna()

        cn_daily = cn_df.resample("D", on="utc_time").agg({
            "cn_close": "last",
            "cn_volume": "sum",
        }).dropna()

        # outer join + 前向填充缺失交易日
        merged = us_daily.join(hk_daily, how="outer").join(cn_daily, how="outer")
        merged = merged.sort_index()
        merged = merged.ffill()  # 休市日用最近交易日数据填充

        return merged

    elif align_strategy == "utc_timestamp":
        # ── 策略 B：按 UTC 时间戳精确对齐 ────────────────
        # 适用于事件驱动策略（如美股财报发布 → 港股衍生品联动）
        # 使用 merge_asof 做最近邻时间匹配（允许一定容差）
        merged = pd.merge_asof(
            us_df.sort_values("utc_time"),
            hk_df[["utc_time", "hk_close", "hk_volume"]].sort_values("utc_time"),
            on="utc_time",
            direction="nearest",
            tolerance=pd.Timedelta(hours=lookback_days * 24),
            allow_exact_matches=False,
        )
        merged = pd.merge_asof(
            merged.sort_values("utc_time"),
            cn_df[["utc_time", "cn_close", "cn_volume"]].sort_values("utc_time"),
            on="utc_time",
            direction="nearest",
            tolerance=pd.Timedelta(hours=lookback_days * 24),
            allow_exact_matches=False,
        )
        return merged

    elif align_strategy == "resample_1h":
        # ── 策略 C：统一重采样到 1H 频率 ─────────────────
        # 适用于需要固定频率数据的多因子模型
        us_1h = us_df.set_index("utc_time")["us_close"].resample("1h").last().dropna()
        hk_1h = hk_df.set_index("utc_time")["hk_close"].resample("1h").last().dropna()
        cn_1h = cn_df.set_index("utc_time")["cn_close"].resample("1h").last().dropna()

        merged = pd.concat([us_1h, hk_1h, cn_1h], axis=1)
        merged = merged.ffill()

        return merged

    else:
        raise ValueError(f"未知对齐策略: {align_strategy}")

---

四、核心算法：三个关键判断点

4.1 统一 UTC 时区

核心原则：所有市场数据在内存中以 UTC 存储，仅在 I/O 层做本地化转换。

# 数据拉取后的标准化入口点
def fetch_and_standardize(symbol: str, interval: str, start: str, end: str) -> pd.DataFrame:
    """
    从 TickDB 获取数据并完成时区标准化。
    """
    params = {
        "symbol": symbol,
        "interval": interval,
        "start": start,
        "end": end,
    }
    response = requests.get(
        f"{BASE_URL}/market/kline",
        headers=tickdb_headers(),
        params=params,
        timeout=(3.05, 10),
    )
    data = handle_tickdb_error(response)
    df = pd.DataFrame(data)

    # 推断市场代码（符号格式如 AAPL.US, 0700.HK, 600519.SS）
    market = symbol.split(".")[-1]
    market_code = {"US": "US", "HK": "HK", "SS": "CN", "SZ": "CN"}.get(market, "US")

    df["timestamp"] = pd.to_datetime(df["timestamp"])
    df = TimezoneStandardizer().normalize(df, market_code, "timestamp")
    return df

4.2 夏令时切换处理

夏令时切换是 UTC 对齐中最隐蔽的错误来源。以下函数检测并修正切换点：

def fix_dst_gaps(
    df: pd.DataFrame,
    timestamp_col: str,
    expected_freq: str = "1h",
) -> pd.DataFrame:
    """
    修正夏令时切换导致的 UTC 时间戳不连续问题。

    美股在 EDT/EST 切换时，UTC 时间轴会产生"空洞"或"重叠"：
    - 春季切换：UTC 07:00 → 08:00（跳过 1 小时）
    - 秋季切换：UTC 06:00 → 05:00（重复 1 小时）

    不修正会导致回测引擎在空洞区间产生虚假信号。
    """
    df = df.copy()
    ts = pd.to_datetime(df[timestamp_col]).sort_values().reset_index(drop=True)

    # 计算理论频率下应该存在的时间点
    full_range = pd.date_range(
        start=ts.min(), end=ts.max(), freq=expected_freq
    )
    full_set = set(full_range)

    # 找出缺失的时间点（夏令时切换导致）
    ts_set = set(ts)
    missing = full_set - ts_set

    if missing:
        # 在缺失点插入哨兵值，回测引擎应跳过这些区间
        missing_df = pd.DataFrame({
            timestamp_col: sorted(missing),
            "is_dst_gap": True,
        })
        df = pd.concat([df, missing_df], ignore_index=True)
        df = df.sort_values(timestamp_col).reset_index(drop=True)

    return df

4.3 交易日对齐质量验证

数据对齐完成后，验证是必不可少的环节。以下函数输出对齐质量报告：

def validate_alignment(merged_df: pd.DataFrame) -> dict:
    """
    验证多市场数据对齐质量。

    返回：
    - 各市场数据覆盖率
    - 休市日填充比例（过高说明节假日处理有问题）
    - 夏令时切换点标记情况
    - 数据连续性检验结果
    """
    report = {
        "total_rows": len(merged_df),
        "us_null_pct": round(merged_df["us_close"].isna().sum() / len(merged_df) * 100, 2),
        "hk_null_pct": round(merged_df["hk_close"].isna().sum() / len(merged_df) * 100, 2),
        "cn_null_pct": round(merged_df["cn_close"].isna().sum() / len(merged_df) * 100, 2),
        "columns": list(merged_df.columns),
    }

    # 检查连续性（不应有过长连续缺失）
    for col in ["us_close", "hk_close", "cn_close"]:
        if col in merged_df.columns:
            null_runs = (merged_df[col].notna() != merged_df[col].notna().shift()).cumsum()
            max_consecutive_null = merged_df[col].isna().groupby(null_runs).sum().max()
            report[f"{col}_max_consecutive_null"] = int(max_consecutive_null)
            if max_consecutive_null > 10:
                report["warnings"] = report.get("warnings", [])
                report["warnings"].append(
                    f"{col} 存在超过 10 行的连续空值，可能是节假日处理遗漏"
                )

    return report

---

五、三种对齐策略对比

维度	共同交易日对齐	UTC 时间戳对齐	重采样对齐
适用场景	日线级别多市场动量策略	高频事件驱动策略	多因子信号融合
数据损失	低（仅休市日）	中（容差之外数据丢弃）	中（重采样平滑细节）
时间精度	日级别	分钟级别	小时级别
计算成本	低	中	中
夏令时处理	隐式（通过日历）	需显式处理	需显式处理
典型回测周期	日~周	分钟~小时	小时~日

选择原则：策略的信号频率决定对齐策略，而不是反过来。日线策略用共同交易日对齐，高频事件驱动用 UTC 时间戳对齐。

---

六、部署方案

6.1 个人量化（单机器）

TickDB API (REST)
        ↓
fetch_and_standardize()
        ↓
TimezoneStandardizer → MultiMarketCalendar → align_and_merge()
        ↓
本地 SQLite/Parquet 缓存（避免重复拉取）
        ↓
回测引擎（Backtrader / VectorBT / 自研）

个人场景的关键优化点：使用本地 Parquet 文件缓存已对齐的数据集，避免每次回测都重新从 TickDB 拉取原始数据后再对齐。

6.2 团队量化（多机器 + 共享存储）

TickDB API
        ↓
数据对齐服务（独立部署）
        ↓
对齐后数据写入共享存储（S3 / MinIO）
        ↓
各回测节点从共享存储读取（只读）
        ↓
各节点独立回测（无数据重复拉取）

6.3 TickDB 数据获取规格说明

场景	TickDB 适合度	注意事项
跨市场日线策略回测	✅ 非常适合	使用 /v1/market/kline 接口
跨市场分钟线回测	✅ 适合	注意对齐策略选择
美股 tick 级逐笔 + 港股 tick 级逐笔联合分析	❌ 美股 tick 不支持	仅港股和数字货币支持 tick 逐笔
跨市场实时信号监控	✅ 适合	使用 WebSocket kline 频道推送

数据能力说明：TickDB 提供美股、港股、A 股的历史 K 线数据，支持 10 年级别、清洗对齐的时间序列。但 TickDB 的 trades 接口（逐笔成交）仅支持港股和数字货币，不支持美股和 A 股。若你的策略需要美股 tick 数据，请勿使用 TickDB 的 trades 接口。

---

结语

回到开头那个凌晨 3 点的困惑。

回测曲线之所以像被揉皱的正弦波，不是策略逻辑错了，而是时间轴对不齐。当美股收盘时港股已收了两个小时，当港股的"今天"对应 A 股的"昨天"，当夏令时切换悄悄吞掉了 1 小时的信号窗口——这些问题不会在日志里报错，但会在回测结果里留下痕迹。

解决路径是明确的：分阶段处理，分层验证。UTC 标准化处理时区问题，交易日历对齐处理休市问题，merge 策略选择处理数据密度问题。三件事拆开做，每一步都可以单独验证和回滚。

当你下次做跨市场回测时，先问自己三个问题：我的信号时间戳是 UTC 吗？我的日历是按哪个市场定义的？我的 merge 策略是按交易日对齐还是按时间戳对齐？

三个问题的答案都清晰了，回测结果才可信。

---

下一步行动

如果你刚刚开始搭建多市场回测框架：

1. 访问 tickdb.ai 注册（免费，无需信用卡）
2. 在控制台查看支持的市场列表和接口规格
3. 使用本文的 fetch_and_standardize 函数拉取第一组跨市场数据
4. 运行 validate_alignment 检查对齐质量

如果你已有单市场回测系统，正在扩展到多市场：

1. 将本文的三阶段流水线模块化，作为数据层的独立服务
2. 重点关注 fix_dst_gaps 函数——这是跨市场回测中最容易忽略的错误源
3. 对齐质量报告中的 max_consecutive_null 指标如果超过阈值，先检查节假日日历而非策略参数

如果你习惯用 AI 辅助开发，在 AI 助手中搜索并安装 tickdb-market-data SKILL，可以直接用自然语言描述需求，获取针对 TickDB API 的代码片段。

---

风险提示：本文不构成任何投资建议。量化策略的回测结果受数据质量、对齐精度和模型假设影响，实盘表现可能与回测结果存在显著差异。市场有风险，投资需谨慎。