行情数据接入与清洗

// CTP 行情回调：每个 tick 都是完整快照
void OnRtnDepthMarketData(CThostFtdcDepthMarketDataField* p) {
    Tick t;
    t.symbol     = p->InstrumentID;
    t.last_price = p->LastPrice;
    t.bid1       = p->BidPrice1;
    t.ask1       = p->AskPrice1;
    t.bid_vol1   = p->BidVolume1;
    t.ask_vol1   = p->AskVolume1;
    t.volume     = p->Volume;          // 累计成交量，非增量
    t.turnover   = p->Turnover;        // 累计成交额
    t.update_ms  = p->UpdateMillisec;
    t.action_day = p->ActionDay;       // 交易日，夜盘跨日关键
    dispatcher_.push(std::move(t));    // 推入无锁队列
}

class OrderBook:
    def __init__(self):
        self.bids = {}   # price -> qty，买单
        self.asks = {}   # price -> qty，卖单
        self.last_seq = 0

    def apply(self, msg):
        if msg.seq != self.last_seq + 1:   # 序号校验：丢包立即触发快照重建
            raise SequenceGap(self.last_seq, msg.seq)
        if msg.type == 'ADD':
            book = self.bids if msg.side == 'B' else self.asks
            book[msg.price] = book.get(msg.price, 0) + msg.qty
        elif msg.type == 'CANCEL':
            book = self.bids if msg.side == 'B' else self.asks
            book[msg.price] = max(0, book.get(msg.price, 0) - msg.qty)
            if book[msg.price] == 0:
                del book[msg.price]
        elif msg.type == 'TRADE':
            book = self.asks if msg.side == 'B' else self.bids
            book[msg.price] = max(0, book.get(msg.price, 0) - msg.qty)
        self.last_seq = msg.seq

    def best_bid(self):
        return max(self.bids) if self.bids else None

    def best_ask(self):
        return min(self.asks) if self.asks else None

// 用硬件时间戳网卡测量行情延迟
struct TimedTick {
    Tick tick;
    uint64_t hw_rx_ns;   // 网卡硬件打戳，纳秒
    uint64_t sw_rx_ns;   // 用户态收到时刻
    uint64_t done_ns;    // 策略处理完成时刻
};

void on_packet(const uint8_t* buf, size_t len, uint64_t hw_ns) {
    TimedTick t;
    t.hw_rx_ns = hw_ns;
    t.sw_rx_ns = now_ns();
    parse(buf, len, t.tick);
    t.done_ns = now_ns();
    metrics_.record("md.total_ns", t.done_ns - t.hw_rx_ns);
}

def clean_tick(t, prev, limits):
    if t.last_price <= 0 or t.last_price > limits.upper:
        return None                       # 价格越界
    if t.bid1 > 0 and t.ask1 > 0 and t.bid1 >= t.ask1:
        return None                       # 买卖倒挂
    if prev and t.ts <= prev.ts:
        t.ts = prev.ts + 1                # 时间倒流，单调化
    if prev and t.volume < prev.volume:
        return None                       # 累计量回退，疑似重启
    return t

def adjust_backward(bars, dividends):
    factor = 1.0   # 后复权：从最早一天开始，逐日累乘复权因子
    for bar in bars:
        if bar.ex_date in dividends:
            d = dividends[bar.ex_date]
            factor *= (bar.close + d.cash) / bar.close
        bar.adj_close = bar.close * factor
    return bars

import pyarrow as pa
import pyarrow.parquet as pq

table = pa.Table.from_pylist(ticks, schema=TICK_SCHEMA)
pq.write_table(
    table,
    f"ticks/{symbol}/{date}.parquet",
    compression="zstd",
    compression_level=3,      # 3 是速度/压缩比的最佳平衡点
    row_group_size=100_000,   # 每 10 万行一个 row group
)

加速回放：把历史 tick 按压缩时间轴快速灌入（用于回测）
实时回放：按原始时间间隔灌入（用于仿真盘）

class ReplayFeed:
    def __init__(self, path, speed=0):
        self.ticks = pq.read_table(path).to_pylist()
        self.speed = speed        # 0 表示全速，1 表示实时

    def run(self, on_tick):
        base_ts = self.ticks[0]['ts']
        t0 = time.time_ns()
        for t in self.ticks:
            if self.speed > 0:
                target = t0 + int((t['ts'] - base_ts) / self.speed)
                while time.time_ns() < target:
                    pass          # 自旋等待，保证时序
            on_tick(t)

class DualFeedMerger:
    def __init__(self):
        self.seen = LRUSet(maxsize=100_000)
        self.primary_ok = True

    def merge(self, t, source):
        key = (t.symbol, t.ts, t.last_price, t.volume)
        if key in self.seen:
            return None                 # 重复，丢弃
        self.seen.add(key)
        return t