180 lines
7.4 KiB
Python
180 lines
7.4 KiB
Python
"""腾讯财经数据源适配器。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional
|
||
from urllib.parse import urlencode
|
||
from urllib.request import Request, urlopen
|
||
|
||
from sidecar.models import KlineBar, Snapshot, parse_date, parse_float
|
||
from sidecar.symbols import normalize_symbol
|
||
|
||
|
||
class TencentFinanceSource:
|
||
"""腾讯财经行情适配器。"""
|
||
|
||
def __init__(self, timeout_seconds: int = 8) -> None:
|
||
"""
|
||
功能说明：创建腾讯财经数据源。
|
||
参数说明：timeout_seconds 为 HTTP 请求超时时间。
|
||
返回值说明：无返回值。
|
||
注意事项：腾讯接口无需 Key，但字段为 GBK 编码和波浪线分隔。
|
||
"""
|
||
self.timeout_seconds = timeout_seconds
|
||
|
||
def fetch_kline(self, symbol: str, period: str, limit: int) -> List[KlineBar]:
|
||
"""
|
||
功能说明：获取腾讯财经 K 线数据。
|
||
参数说明：symbol 为股票或指数代码，period 为周期，limit 为最大条数。
|
||
返回值说明：返回统一 KlineBar 列表。
|
||
注意事项：当前仅支持 day/week/month，用于补充指数历史 K 线。
|
||
"""
|
||
period_map = {"day": "day", "week": "week", "month": "month"}
|
||
tencent_period = period_map.get(period)
|
||
if tencent_period is None:
|
||
raise ValueError("腾讯财经 K 线暂不支持周期: %s" % period)
|
||
normalized = normalize_symbol(symbol)
|
||
url = "https://web.ifzq.gtimg.cn/appstock/app/kline/kline?" + urlencode(
|
||
{"param": "%s,%s,,,%d" % (normalized, tencent_period, limit)}
|
||
)
|
||
request = Request(url, headers={"User-Agent": "Mozilla/5.0 TortoiseSidecar/0.1"})
|
||
with urlopen(request, timeout=self.timeout_seconds) as response:
|
||
payload = response.read().decode("utf-8", errors="ignore")
|
||
return kline_from_tencent_payload(normalized, period, payload)
|
||
|
||
def fetch_snapshot(self, symbol: str) -> Snapshot:
|
||
"""
|
||
功能说明：获取腾讯财经实时行情快照。
|
||
参数说明：symbol 为股票代码。
|
||
返回值说明：返回统一 Snapshot 模型。
|
||
注意事项：PB 使用索引 46，PE_TTM 使用索引 39，避免常见错误映射。
|
||
"""
|
||
normalized = normalize_symbol(symbol)
|
||
query = urlencode({"q": normalized})
|
||
request = Request(
|
||
"https://qt.gtimg.cn/?" + query,
|
||
headers={"User-Agent": "Mozilla/5.0 TortoiseSidecar/0.1"},
|
||
)
|
||
with urlopen(request, timeout=self.timeout_seconds) as response:
|
||
text = response.read().decode("gbk", errors="ignore")
|
||
fields = parse_tencent_payload(text)
|
||
return snapshot_from_tencent_fields(normalized, fields)
|
||
|
||
|
||
def parse_tencent_payload(payload: str) -> List[str]:
|
||
"""
|
||
功能说明：解析腾讯财经原始响应。
|
||
参数说明：payload 为形如 v_sh600000="..." 的响应文本。
|
||
返回值说明：返回按波浪线拆分后的字段列表。
|
||
注意事项：响应为空或格式异常时抛出 ValueError。
|
||
"""
|
||
if '="' not in payload:
|
||
raise ValueError("腾讯财经响应格式异常")
|
||
content = payload.split('="', 1)[1].split('"', 1)[0]
|
||
fields = content.split("~")
|
||
if len(fields) < 53:
|
||
raise ValueError("腾讯财经响应字段不足")
|
||
return fields
|
||
|
||
|
||
def kline_from_tencent_payload(symbol: str, period: str, payload: str) -> List[KlineBar]:
|
||
"""
|
||
功能说明：把腾讯财经 K 线 JSON 转换为统一 K 线列表。
|
||
参数说明：symbol 为标准股票代码，period 为统一周期，payload 为腾讯 K 线 JSON 文本。
|
||
返回值说明：返回 KlineBar 列表。
|
||
注意事项：腾讯字段顺序为 日期、开盘、收盘、最高、最低、成交量，部分响应可能附带成交额。
|
||
"""
|
||
data = json.loads(payload)
|
||
if data.get("code") != 0:
|
||
raise ValueError("腾讯财经 K 线响应失败: %s" % data.get("msg"))
|
||
symbol_data = data.get("data", {}).get(symbol, {})
|
||
rows = symbol_data.get(period) or symbol_data.get("qfq%s" % period) or []
|
||
bars = []
|
||
for row in rows:
|
||
trade_date = parse_date(_row_field(row, 0))
|
||
if trade_date is None:
|
||
continue
|
||
bars.append(
|
||
KlineBar(
|
||
symbol=symbol,
|
||
period=period,
|
||
trade_date=trade_date,
|
||
open=parse_float(_row_field(row, 1)),
|
||
close=parse_float(_row_field(row, 2)),
|
||
high=parse_float(_row_field(row, 3)),
|
||
low=parse_float(_row_field(row, 4)),
|
||
volume=parse_float(_row_field(row, 5)),
|
||
amount=parse_float(_row_field(row, 6)),
|
||
source="tencent",
|
||
)
|
||
)
|
||
return bars
|
||
|
||
|
||
def snapshot_from_tencent_fields(symbol: str, fields: List[str]) -> Snapshot:
|
||
"""
|
||
功能说明：把腾讯财经字段转换为统一快照。
|
||
参数说明：symbol 为标准股票代码，fields 为腾讯响应字段列表。
|
||
返回值说明：返回 Snapshot 模型。
|
||
注意事项：字段索引按项目文档校准，43 是振幅不是 PB。
|
||
"""
|
||
return Snapshot(
|
||
symbol=symbol,
|
||
name=_field(fields, 1) or None,
|
||
trade_time=_parse_trade_time(_field(fields, 30)),
|
||
price=parse_float(_field(fields, 3)),
|
||
previous_close=parse_float(_field(fields, 4)),
|
||
open=parse_float(_field(fields, 5)),
|
||
high=parse_float(_field(fields, 33)),
|
||
low=parse_float(_field(fields, 34)),
|
||
volume=parse_float(_field(fields, 6)),
|
||
amount=parse_float(_field(fields, 37)),
|
||
change=parse_float(_field(fields, 31)),
|
||
change_percent=parse_float(_field(fields, 32)),
|
||
turnover_rate=parse_float(_field(fields, 38)),
|
||
pe_ttm=parse_float(_field(fields, 39)),
|
||
pe_static=parse_float(_field(fields, 52)),
|
||
pb=parse_float(_field(fields, 46)),
|
||
market_cap=parse_float(_field(fields, 44)),
|
||
float_market_cap=parse_float(_field(fields, 45)),
|
||
limit_up=parse_float(_field(fields, 47)),
|
||
limit_down=parse_float(_field(fields, 48)),
|
||
source="tencent",
|
||
)
|
||
|
||
|
||
def _field(fields: List[str], index: int) -> str:
|
||
"""
|
||
功能说明：安全读取腾讯字段。
|
||
参数说明：fields 为字段列表，index 为字段索引。
|
||
返回值说明：字段存在返回原始字符串，不存在返回空字符串。
|
||
注意事项：腾讯字段偶发缺失时可避免 IndexError。
|
||
"""
|
||
return fields[index] if index < len(fields) else ""
|
||
|
||
|
||
def _row_field(row: List[Any], index: int) -> Any:
|
||
"""
|
||
功能说明：安全读取腾讯 K 线行字段。
|
||
参数说明：row 为腾讯 K 线数组，index 为字段索引。
|
||
返回值说明：字段存在返回原始值，不存在返回 None。
|
||
注意事项：不同标的可能缺少成交额字段。
|
||
"""
|
||
return row[index] if index < len(row) else None
|
||
|
||
|
||
def _parse_trade_time(value: str) -> Optional[datetime]:
|
||
"""
|
||
功能说明：解析腾讯交易时间字段。
|
||
参数说明：value 为 YYYYMMDDHHMMSS 格式字符串。
|
||
返回值说明：解析成功返回 datetime，空值或异常返回 None。
|
||
注意事项：返回值不带时区，表示交易所本地时间。
|
||
"""
|
||
if not value:
|
||
return None
|
||
try:
|
||
return datetime.strptime(value[:14], "%Y%m%d%H%M%S")
|
||
except ValueError:
|
||
return None |