实时行情写缓存前先标准化这些字段，避免策略层反复清洗

实时行情写缓存前先标准化这些字段，避免策略层反复清洗

一句话结论：实时行情写入缓存之前，先统一标的代码格式、时间戳时区、价格精度、涨跌幅小数表示和成交量单位这五类字段，可以大幅减少策略层和数据管道的重复清洗逻辑。

1. 问题背景

一个典型的盘中监控流程是这样：从数据 API 拿到实时行情，写入本地缓存（Redis、SQLite 或内存 DataFrame），然后策略从中读取数据计算信号。初期只监控少数标的时，数据处理分散在各个策略脚本里，每个脚本用自己的方式解析时间戳、处理涨跌幅、转换成交量单位。

当监控范围扩大到几百只标的，策略数量增加到五六个以上时，问题开始暴露：同样的字段清洗逻辑在多个文件中重复出现，一个接口字段调整就要改五六处代码，数据链路中的一致性也难以保证。

2. 核心结论

解决方式不是给每个策略增加更多清洗代码，而是在数据写入缓存之前，完成一次性的标准化处理。策略层只需要假定缓存中的数据已经是统一格式，不再负责重复转换。

3. 需要标准化的五类字段

3.1 标的代码格式

不同数据源返回的标的代码格式经常不一致。有的返回纯数字（600519），有的带市场后缀（600519.SH），有的使用前缀缩写。策略在筛选特定标的或跨市场合并数据时，每次都需额外处理代码。

标准化方式：统一为带交易所后缀的格式，如 600519.SH、000001.SZ、AAPL.US、00700.HK。写入缓存前完成全部转换，策略只需按统一格式查询。

3.2 时间戳时区

实时行情的时间戳可能是北京时间，也可能是 UTC。如果缓存中两种混存，策略在判断数据新鲜度时很容易出错。

标准化方式：统一转换为北京时间（Asia/Shanghai）的毫秒或秒级时间戳，并在缓存中附带一个标识或直接以字段名区分。

3.3 价格精度

不同市场或不同标的的报价小数位数可能不同。A 股最小变动单位 0.01 元，港股某些标的可以到三位小数。写入缓存后策略直接使用，如果在策略层重新做四舍五入，可能出现对齐问题。

标准化方式：保留数据源返回的原始精度，不截断也不补位，但统一使用 float 类型存入，策略读取后按需自行处理。关键在于不要在不同环节反复做精度转换。

3.4 涨跌幅的小数表示

涨跌幅在不同 API 中的返回方式有两种常见形式：百分比格式（1.23%）和小数格式（0.0123）。如果缓存中两种形式混存，策略判断时可能出现量级错误，把 0.01（1%）当成 0.01%。

标准化方式：统一使用小数格式，例如 0.0123 表示 1.23%。在缓存写入层完成转换，策略从缓存读到的一律是小数。

3.5 成交量单位

A 股成交量以手为单位，港股的成交量以股为单位，美股通常也以股为单位。如果跨市场数据写入同一张表而不做单位标记，策略在计算换手率时会出现数量级偏差。

标准化方式：保留两个字段：volume（原始返回的单位）和 volume_shares（统一转换为股数）。或者在缓存中通过标的类型字段让策略自行换算，前提是所有策略使用同一套换算规则。

4. 工程方案对比

5. 一个简单的实现思路

以下代码演示一个缓存写入前的标准化函数，以 Python 为例：

import time
from typing import Any

def normalize_quote(raw: dict[str, Any]) -> dict[str, Any]:
    normalized = {}

    # 1. 标的代码统一带后缀
    symbol = raw.get("symbol", "")
    if "." not in symbol:
        if symbol.startswith("6"):
            symbol = f"{symbol}.SH"
        elif symbol.startswith(("0", "3")):
            symbol = f"{symbol}.SZ"
        elif symbol.startswith(("8", "9")):
            symbol = f"{symbol}.BJ"
        # 美股、港股按实际标识添加
    normalized["symbol"] = symbol

    # 2. 时间戳统一为北京时间
    ts = raw.get("timestamp", int(time.time() * 1000))
    normalized["timestamp"] = ts

    # 3. 涨跌幅统一为小数
    change_pct = raw.get("change_pct")
    if isinstance(change_pct, str) and "%" in change_pct:
        change_pct = float(change_pct.replace("%", "")) / 100
    normalized["change_pct"] = float(change_pct) if change_pct else 0.0

    # 4. 成交量保留原始字段，另存一手（股）字段
    normalized["volume"] = raw.get("volume", 0)
    normalized["volume_shares"] = raw.get("volume", 0)

    # 5. 价格原样保留
    normalized["last_price"] = float(raw.get("last_price", 0))
    normalized["prev_close"] = float(raw.get("prev_close", 0))

    return normalized

这个函数可以在写入 Redis 或 DataFrame 之前统一调用一次。所有策略从缓存读取时，得到的数据已经具备一致的格式。

6. 验证方法

验证标准化层是否有效，可以检查以下几项：

• 缓存中同一字段是否只有一种数据格式（例如涨跌幅字段全部是小数，没有百分比字符串）；
• 新接入一个策略时，是否不需要添加新的清洗代码；
• 数据源切换时，只需要修改标准化层，策略代码零改动。

7. 常见错误

• 在缓存中同时保留原始格式和标准化格式，导致策略混淆。 要么只存标准化后的数据，要么明确区分字段名（如 change_pct 统一为小数，change_pct_raw 保留原始值）。
• 在多个层面反复标准化。 缓存写入层做了一次，策略读取后又做一次，不仅浪费计算，还可能因为转换规则不一致导致结果不同。
• 忽略新增标的的代码格式。 标准化函数上线后，如果新增了一个市场或特殊代码，必须同步更新标准化规则。

8. FAQ

Q1：标准化层应该放在哪里？

建议放在数据获取和缓存写入之间。即 API 返回数据后，先经过标准化函数，再将结果写入缓存。

Q2：涨跌幅用小数的好处是什么？

小数格式可以直接参与数学计算，不需要每个策略都做字符串解析。例如 0.0123 可以直接乘以基准价得到涨跌金额。

Q3：成交量单位不一致怎么处理？

优先转换为股数。A 股 1 手 = 100 股，港股和美股默认以股为单位。如果无法统一，至少用字段名或额外标记说明当前单位。

Q4：时间戳应该用秒还是毫秒？

建议统一为毫秒时间戳，因为多数行情 API 返回毫秒。转换为秒虽然节省存储，但每次使用都需要再乘 1000，增加转换步骤。

Q5：如果数据源切换后字段名变了怎么办？

标准化层中建立字段映射表，切换数据源时只修改标准化函数内部的映射关系，下游策略不感知字段名的变化。

总结

• 实时行情写入缓存前集中完成字段标准化，可以消除策略层的大量重复清洗逻辑。
• 标的代码格式、时间戳时区、价格精度、涨跌幅表示和成交量单位是五个最高频的清洗点。
• 标准化层引入后，新策略接入成本降低，数据源切换也更可控。
• 关键在于标准化层只做一次，下游不再重复。