盯盘程序别到处判断市场

原创

用户9138916

发布于 2026-09-17 18:05:41

1480

很多盯盘程序一开始只看几只 A 股，代码里写几句 if symbol.endswith(".SH") 也能跑。等到需求变成同时看沪深京股票、场内 ETF，甚至后续还要接入港股、美股时，市场判断就会散落在行情拉取、交易日过滤、涨跌幅计算、告警规则和页面展示里。

更稳妥的做法是：不要让业务代码到处判断“这是股票还是 ETF、这是哪个交易所”。应该把这些差异收敛到两层：一层是标的元数据，负责描述标的属于什么市场、什么品种、使用什么交易日历；另一层是行情适配器，负责把不同数据源转换成统一的数据结构。策略和告警只面对统一的 Snapshot，不直接关心底层市场差异。

下面用一个最小 Python 结构说明如何组织这类盯盘代码。示例重点不是某个行情源，而是边界划分：哪些判断应该集中管理，哪些判断不应该进入策略逻辑。

问题不在 if，而在 if 出现的位置

盯盘程序中常见的市场差异包括：

标的代码格式不同，例如 600519.SH、000001.SZ、510300.SH；
股票和 ETF 的基础字段相似，但业务含义不同；
不同市场可能有不同交易日历、交易时段和涨跌停规则；
部分字段可能只对某类标的有意义；
告警规则通常相同，但阈值、白名单和展示名称可能来自配置。

如果这些差异被写在每个业务函数里，代码很快会变成这样：

def check_alert(symbol, quote):
    if symbol.endswith(".SH") or symbol.endswith(".SZ"):
        # A 股逻辑
        pass

    if symbol.startswith("51") or symbol.startswith("15"):
        # ETF 逻辑
        pass

这种写法的问题不是 if 本身，而是判断的位置错了。市场识别是数据建模问题，不应该散落到告警、展示、计算和任务调度里。

先定义统一的领域对象

建议先把“标的”和“行情快照”定义成稳定结构。后续无论数据来自 REST API、SDK、WebSocket，还是本地缓存，都先转换成这些对象。

运行环境可以使用：

Python 3.10+
pandas 2.x（如果需要批量表格处理）
标准库 dataclasses、enum、datetime

from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Optional


class AssetType(str, Enum):
    STOCK = "stock"
    ETF = "etf"


class Region(str, Enum):
    CN = "CN"


@dataclass(frozen=True)
class Instrument:
    symbol: str          # 统一代码，例如 600519.SH、510300.SH
    name: str
    region: Region
    asset_type: AssetType
    exchange: str        # SH / SZ / BJ
    calendar: str        # 例如 XSHG、XSHE，也可以按业务统一成 CN_A


@dataclass(frozen=True)
class Snapshot:
    symbol: str
    ts: datetime
    last_price: float
    prev_close: float
    volume: Optional[float] = None
    amount: Optional[float] = None

    @property
    def change_pct(self) -> float:
        if self.prev_close <= 0:
            return 0.0
        return self.last_price / self.prev_close - 1.0

这里有两个关键点。

第一，Instrument 负责表达“它是什么”。股票、ETF、交易所、交易日历都放在这里。

第二，Snapshot 负责表达“现在行情是多少”。涨跌幅这种通用计算可以放在快照对象里，但不要在里面写“如果是 ETF 就如何，如果是股票就如何”的复杂业务分支。

用注册表集中管理标的差异

对中小规模盯盘程序来说，最简单的办法是维护一个标的注册表。它可以来自 YAML、数据库、接口返回结果，也可以先用 Python 字典表达。

INSTRUMENTS = {
    "600519.SH": Instrument(
        symbol="600519.SH",
        name="贵州茅台",
        region=Region.CN,
        asset_type=AssetType.STOCK,
        exchange="SH",
        calendar="CN_A",
    ),
    "000001.SZ": Instrument(
        symbol="000001.SZ",
        name="平安银行",
        region=Region.CN,
        asset_type=AssetType.STOCK,
        exchange="SZ",
        calendar="CN_A",
    ),
    "510300.SH": Instrument(
        symbol="510300.SH",
        name="沪深300ETF",
        region=Region.CN,
        asset_type=AssetType.ETF,
        exchange="SH",
        calendar="CN_A",
    ),
}


def get_instrument(symbol: str) -> Instrument:
    try:
        return INSTRUMENTS[symbol]
    except KeyError as exc:
        raise ValueError(f"unknown instrument: {symbol}") from exc

这样做以后，市场判断只在注册表生成或维护时出现。告警代码拿到的已经是结构化信息，不需要靠字符串猜测。

如果使用第三方数据服务，也建议优先选用带市场后缀的统一代码格式。例如 QuantDash 官方文档中的标的格式为 {代码}.{交易所后缀}，A 股可使用 600519.SH、000001.SZ，ETF 可按对应交易所后缀组织；但无论使用哪家数据源，都应先进入自己的 Instrument 和 Snapshot 模型，而不是让数据源字段直接进入业务层。

行情适配器只负责转换，不负责告警

行情源通常有自己的字段名。有的返回 last_price，有的返回 price；有的把涨跌幅写成小数，有的写成百分数；有的时间戳是毫秒，有的是秒。

这些差异应该放进适配器。

from abc import ABC, abstractmethod
from datetime import timezone


class QuoteProvider(ABC):
    @abstractmethod
    def get_snapshots(self, symbols: list[str]) -> dict[str, Snapshot]:
        raise NotImplementedError


class DemoQuoteProvider(QuoteProvider):
    """示例适配器：把外部原始行情转换为 Snapshot。"""

    def get_snapshots(self, symbols: list[str]) -> dict[str, Snapshot]:
        raw_rows = self._fetch_raw_quotes(symbols)
        return {row["symbol"]: self._to_snapshot(row) for row in raw_rows}

    def _fetch_raw_quotes(self, symbols: list[str]) -> list[dict]:
        # 实际项目中这里调用行情 API、SDK 或读取缓存。
        # 示例不伪造接口输出，只保留字段转换结构。
        raise NotImplementedError

    def _to_snapshot(self, row: dict) -> Snapshot:
        ts = datetime.fromtimestamp(row["timestamp"] / 1000, tz=timezone.utc)
        return Snapshot(
            symbol=row["symbol"],
            ts=ts,
            last_price=float(row["last_price"]),
            prev_close=float(row["prev_close"]),
            volume=float(row["volume"]) if row.get("volume") is not None else None,
            amount=float(row["amount"]) if row.get("amount") is not None else None,
        )

适配器里可以有字段判断、时间单位处理和空值修正，但不应该写“涨幅超过多少就报警”。这样后续更换数据源时，只改适配器，不改策略层。

告警规则依赖类型，不依赖代码字符串

有了 Instrument 和 Snapshot，告警规则就可以写得更干净。

@dataclass(frozen=True)
class AlertRule:
    asset_type: AssetType
    change_threshold: float


DEFAULT_RULES = {
    AssetType.STOCK: AlertRule(
        asset_type=AssetType.STOCK,
        change_threshold=0.03,
    ),
    AssetType.ETF: AlertRule(
        asset_type=AssetType.ETF,
        change_threshold=0.015,
    ),
}


def should_alert(instrument: Instrument, snapshot: Snapshot) -> bool:
    rule = DEFAULT_RULES[instrument.asset_type]
    return abs(snapshot.change_pct) >= rule.change_threshold

这里仍然区分股票和 ETF，但区分依据来自 instrument.asset_type，而不是 symbol.startswith()。这就是代码组织上的核心变化：业务层可以使用类型信息，但不要自己推断类型。

如果后续增加债券 ETF、跨境 ETF、可转债或港股，应该扩展元数据和规则配置，而不是继续在业务函数里叠加字符串判断。

批量任务按能力分组，而不是按代码分支分组

盯盘程序通常会批量请求行情。一个常见误区是直接按市场写多个分支：

# 不推荐
sh_symbols = [s for s in symbols if s.endswith(".SH")]
sz_symbols = [s for s in symbols if s.endswith(".SZ")]

更好的办法是按“数据源能力”或“调度能力”分组。例如某个行情接口一次可以同时查 A 股和 ETF，那就不需要拆；如果某个接口要求不同市场走不同端点，也应该由 provider 自己拆。

class WatchEngine:
    def __init__(self, provider: QuoteProvider):
        self.provider = provider

    def run_once(self, symbols: list[str]) -> list[str]:
        instruments = {symbol: get_instrument(symbol) for symbol in symbols}
        snapshots = self.provider.get_snapshots(symbols)

        messages = []
        for symbol, instrument in instruments.items():
            snapshot = snapshots.get(symbol)
            if snapshot is None:
                continue

            if should_alert(instrument, snapshot):
                messages.append(format_alert(instrument, snapshot))

        return messages


def format_alert(instrument: Instrument, snapshot: Snapshot) -> str:
    pct = snapshot.change_pct * 100
    return (
        f"{instrument.name}({instrument.symbol}) "
        f"最新价 {snapshot.last_price:.3f}，"
        f"较昨收 {pct:.2f}%"
    )

WatchEngine 不知道底层接口怎么拆请求，也不知道股票和 ETF 的代码前缀规则。它只做三件事：加载元数据、获取统一快照、执行告警规则。

配置比硬编码更适合长期维护

当盯盘标的增多时，可以把注册表和规则放到配置文件中。示例 YAML：

instruments:
  - symbol: 600519.SH
    name: 贵州茅台
    region: CN
    asset_type: stock
    exchange: SH
    calendar: CN_A

  - symbol: 510300.SH
    name: 沪深300ETF
    region: CN
    asset_type: etf
    exchange: SH
    calendar: CN_A

rules:
  stock:
    change_threshold: 0.03
  etf:
    change_threshold: 0.015

配置化以后，新增标的大多数时候只是改配置，不需要改代码。更重要的是，代码审查时可以很容易判断：哪些改动是业务规则变化，哪些改动是程序结构变化。

时间、排序和字段要在入口处校验

盯盘程序看似只处理实时行情，但实际故障常出现在数据边界：

时间戳单位是秒还是毫秒；
时间戳时区是 UTC、交易所本地时间，还是服务端时间；
prev_close 是否缺失或为 0；
行情是否延迟；
同一个标的是否出现重复快照；
股票停牌或 ETF 临时异常时字段是否为空。

这些校验也应该放在适配器或数据入口，而不是在每条告警规则里重复写。

def validate_snapshot(snapshot: Snapshot) -> None:
    if snapshot.last_price < 0:
        raise ValueError(f"negative price: {snapshot.symbol}")
    if snapshot.prev_close < 0:
        raise ValueError(f"negative prev_close: {snapshot.symbol}")
    if snapshot.ts.tzinfo is None:
        raise ValueError(f"timezone required: {snapshot.symbol}")

校验规则不一定都要抛异常。对盯盘系统来说，可以按严重程度分成三类：

阻断型：代码未知、价格为负、时间字段不可解析；
降级型：成交额缺失、扩展字段缺失；
观测型：行情时间落后、连续多次未更新。

阻断型问题不应进入策略层，降级型问题可以进入但要有默认行为，观测型问题应该进入日志和监控。

一个可落地的目录结构

中小型 Python 项目可以从下面的结构开始：

watcher/
  domain.py          # Instrument、Snapshot、枚举
  registry.py        # 标的注册表加载与校验
  providers/
    base.py          # QuoteProvider 接口
    demo.py          # 具体数据源适配器
  rules.py           # 告警规则
  engine.py          # WatchEngine
  config/
    instruments.yaml
    rules.yaml

各模块职责应保持清晰：

domain.py 不依赖外部 API；
providers/ 可以依赖数据源 SDK 或 HTTP 客户端；
rules.py 不直接调用行情接口；
engine.py 负责编排，不负责解析市场代码；
config/ 承载标的和阈值变化。

这种结构不复杂，但能有效避免“市场判断蔓延”。

什么时候仍然需要市场分支

并不是所有分支都要消灭。市场差异真实存在，只是应该放在合适的位置。

适合出现市场分支的位置包括：

标的注册表生成器；
数据源适配器；
交易日历服务；
风控或告警规则配置；
字段兼容层。

不适合出现市场分支的位置包括：

页面展示函数；
普通涨跌幅计算；
通用告警引擎；
任务调度主循环；
日志打印和消息格式化。

判断标准很简单：如果某段代码的职责不是“识别或适配市场差异”，它就不应该自己猜市场。

总结

盯盘程序同时处理 A 股和 ETF 时，代码组织的关键不是把所有 if 都删掉，而是把市场差异集中到边界层。

推荐的拆法是：

用 Instrument 描述标的类型、市场、交易所和日历；
用 Snapshot 统一行情快照字段；
用 QuoteProvider 适配不同数据源；
用配置管理股票、ETF 的规则差异；
让告警和展示逻辑只依赖统一对象，不直接解析代码字符串。

这样写的好处是，今天同时看 A 股和 ETF，明天增加新的标的类型或行情源，主要改动也会集中在元数据、适配器和规则配置里，而不是在整个项目里搜索 if market == ...。

FAQ

A 股和 ETF 可以共用同一个 Snapshot 吗？

可以。只要盯盘逻辑需要的是最新价、昨收、成交量、成交额和时间戳这类通用字段，就可以共用。不同品种的业务差异应放在 Instrument 和规则配置中。

ETF 是否一定要单独写一套告警逻辑？

不一定。如果告警只看涨跌幅、价格和成交量，可以复用同一套规则引擎，只在阈值配置上区分股票和 ETF。只有当 ETF 特有字段进入判断条件时，才需要扩展规则模型。

直接用代码前缀判断 ETF 为什么不好？

代码前缀属于数据识别细节，容易随着市场、交易所和数据源变化而失效。更可靠的方式是在标的注册表中显式保存 asset_type，业务层直接使用这个字段。

多个行情源字段不一致怎么办？

每个行情源各写一个适配器，把原始字段转换成统一的 Snapshot。不要让业务层同时兼容多套字段名，否则数据源越多，策略和告警代码越难维护。

后续增加港股或美股需要重写吗？

如果前期已经把市场差异收敛到元数据、交易日历和适配器层，通常不需要重写主流程。需要新增的是标的类型、交易日历、数据源适配和相应规则配置。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python