从多个行情供应商迁移到统一 API：双跑期、字段映射与切换验收

# parity_check.py — compare two vendor snapshots field by field
# Prices move between calls, so exact equality is the wrong test.
TOL = {"last": 0.005, "bid": 0.005, "ask": 0.005, "volume": 0.05}
MAX_LAG_MS = 5_000
FIELDS = ("last", "bid", "ask", "volume", "timestampMs")


def _rel(a, b):
    return abs(a - b) / (max(abs(a), abs(b)) or 1.0)


def compare(code, old, new):
    out = []
    if bool(old.get("delayed")) != bool(new.get("delayed")):
        out.append((code, "delayed", "flag", old.get("delayed"), new.get("delayed")))
    for f in FIELDS:
        a, b = old.get(f), new.get(f)
        if f == "timestampMs":
            if a and b and abs(a - b) > MAX_LAG_MS:
                out.append((code, f, "lag", a, b))
            continue
        if a is None or b is None:
            if a != b:  # one side missing is a mapping bug, not volatility
                out.append((code, f, "missing", a, b))
            continue
        if _rel(a, b) > TOL.get(f, 0.005):
            out.append((code, f, "value", a, b))
    return out


def report(pairs):
    diffs = [d for code, old, new in pairs for d in compare(code, old, new)]
    for code, field, kind, a, b in diffs:
        print(f"{code:<10} {field:<12} {kind:<8} {a} -> {b}")
    return diffs


if __name__ == "__main__":
    report([
        ("EURUSD", {"last": 1.0850, "bid": 1.0849, "ask": 1.0851, "volume": 1200,
                    "timestampMs": 1_757_000_000_000, "delayed": False},
                   {"last": 1.0851, "bid": 1.0850, "ask": 1.0852, "volume": 1210,
                    "timestampMs": 1_757_000_002_000, "delayed": False}),
        ("XAUUSD", {"last": 2410.5, "bid": None, "ask": None, "volume": 300,
                    "timestampMs": 1_757_000_000_000, "delayed": True},
                   {"last": 2410.9, "bid": 2410.4, "ask": 2410.6, "volume": 302,
                    "timestampMs": 1_757_000_000_500, "delayed": False}),
    ])

# cutover.py — shift consumers, not endpoints
import hashlib

CONSUMERS = {
    "web-terminal": {"pct": 100, "fallback": "legacy"},
    "risk-monitor": {"pct": 25, "fallback": "legacy"},
    "batch-backfill": {"pct": 0, "fallback": "legacy"},
}


def _bucket(consumer, symbol):
    # Deterministic: one consumer and symbol must stay on one side
    # for the whole window, or a dashboard mixes two vendors.
    key = f"{consumer}:{symbol}".encode()
    return int(hashlib.sha256(key).hexdigest()[:8], 16) % 100


def route(consumer, symbol):
    rule = CONSUMERS.get(consumer)
    if rule is None:
        return "legacy"  # an unknown consumer never gets the new feed
    return "unified" if _bucket(consumer, symbol) < rule["pct"] else rule["fallback"]


def rollback(consumer):
    """One line puts a consumer back on the old feed, no redeploy."""
    CONSUMERS[consumer]["pct"] = 0


if __name__ == "__main__":
    for c in CONSUMERS:
        hits = sum(route(c, f"SYM{i}") == "unified" for i in range(200))
        print(f"{c:<16} pct={CONSUMERS[c]['pct']:>3} routed_new={hits}/200")
    rollback("risk-monitor")
    print("after rollback:", route("risk-monitor", "SYM7"))