"""统一数据模型与序列化工具。""" from __future__ import annotations from dataclasses import asdict, dataclass from datetime import date, datetime from typing import Any, Dict, Optional @dataclass(frozen=True) class KlineBar: """统一 K 线数据。""" symbol: str period: str trade_date: date open: Optional[float] high: Optional[float] low: Optional[float] close: Optional[float] volume: Optional[float] amount: Optional[float] source: str @dataclass(frozen=True) class Snapshot: """统一行情快照数据。""" symbol: str name: Optional[str] trade_time: Optional[datetime] price: Optional[float] previous_close: Optional[float] open: Optional[float] high: Optional[float] low: Optional[float] volume: Optional[float] amount: Optional[float] change: Optional[float] change_percent: Optional[float] turnover_rate: Optional[float] pe_ttm: Optional[float] pe_static: Optional[float] pb: Optional[float] market_cap: Optional[float] float_market_cap: Optional[float] limit_up: Optional[float] limit_down: Optional[float] source: str def to_dict(value: Any) -> Dict[str, Any]: """ 功能说明:把 dataclass 模型转换为可 JSON 序列化的字典。 参数说明:value 为 dataclass 实例。 返回值说明:返回字段字典。 注意事项:日期时间对象交给 FastAPI 编码。 """ return asdict(value) def parse_float(value: Any) -> Optional[float]: """ 功能说明:宽松解析浮点数。 参数说明:value 为字符串、数字或空值。 返回值说明:解析成功返回 float,空值或非法值返回 None。 注意事项:腾讯财经空字段、破折号等非数字值会被视为 None。 """ if value is None: return None text = str(value).strip() if text in {"", "-", "--", "None", "nan"}: return None try: return float(text) except ValueError: return None def parse_date(value: Any) -> Optional[date]: """ 功能说明:解析常见日期字段。 参数说明:value 为 date、datetime 或日期字符串。 返回值说明:解析成功返回 date,失败返回 None。 注意事项:支持 YYYY-MM-DD 与 YYYYMMDD 两种主要格式。 """ if isinstance(value, datetime): return value.date() if isinstance(value, date): return value text = str(value).strip() for fmt, size in (("%Y-%m-%d", 10), ("%Y%m%d", 8)): try: return datetime.strptime(text[:size], fmt).date() except ValueError: continue return None