272 lines
10 KiB
Python
272 lines
10 KiB
Python
"""腾讯财经 API 客户端"""
|
||
|
||
import httpx
|
||
import re
|
||
from urllib.parse import quote
|
||
from typing import Optional, List, Dict
|
||
|
||
# 根据代码推断市场前缀（腾讯格式）
|
||
def get_market_prefix(code: str) -> str:
|
||
if code.startswith("688") or code.startswith("60"):
|
||
return "sh"
|
||
if code.startswith("920") or code.startswith("8") or code.startswith("4"):
|
||
return "bj"
|
||
return "sz"
|
||
|
||
|
||
# 解码 \u 转义的 unicode 字符串
|
||
def decode_unicode(s: str) -> str:
|
||
try:
|
||
return s.encode("utf-8").decode("unicode_escape")
|
||
except Exception:
|
||
return s
|
||
|
||
|
||
# 解析腾讯报价文本格式（按 ~ 分隔）
|
||
def parse_tencent_data(text: str):
|
||
eq_idx = text.index('="')
|
||
if eq_idx == -1:
|
||
return None
|
||
start = eq_idx + 2
|
||
end = text.rindex('"')
|
||
if end <= start:
|
||
return None
|
||
content = text[start:end]
|
||
if not content:
|
||
return None
|
||
return content.split("~")
|
||
|
||
|
||
async def search_stock(keyword: str) -> List[dict]:
|
||
"""腾讯智能搜索：支持名称/代码/拼音模糊匹配"""
|
||
url = f"https://smartbox.gtimg.cn/s3/?t=all&q={quote(keyword, safe='')}&v=2"
|
||
headers = {
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
||
"Referer": "https://stockapp.finance.qq.com/",
|
||
}
|
||
results = []
|
||
async with httpx.AsyncClient() as client:
|
||
try:
|
||
resp = await client.get(url, headers=headers, timeout=10)
|
||
if resp.status_code != 200:
|
||
return results
|
||
text = resp.text
|
||
eq_idx = text.index('="')
|
||
if eq_idx == -1:
|
||
return results
|
||
start = eq_idx + 2
|
||
end = text.rindex('"')
|
||
if end <= start:
|
||
return results
|
||
content = text[start:end]
|
||
if not content or content == "N":
|
||
return results
|
||
records = content.split("^")
|
||
for record in records:
|
||
if not record:
|
||
continue
|
||
parts = record.split("~")
|
||
if len(parts) < 5:
|
||
continue
|
||
market = parts[0]
|
||
code = parts[1]
|
||
name_raw = parts[2]
|
||
typ = parts[4]
|
||
is_sh_a = market == "sh" and (typ in ("GP-A", "GP-A-KCB"))
|
||
is_sz_a = market == "sz" and (typ in ("GP-A", "GP-A-CYB"))
|
||
is_bj_a = market == "bj" and (typ in ("GP-A", "GP-A-BJB"))
|
||
if (is_sh_a or is_sz_a or is_bj_a) and re.match(r"^\d{6}$", code):
|
||
name = decode_unicode(name_raw)
|
||
if name:
|
||
results.append({
|
||
"code": code,
|
||
"name": name,
|
||
"market": market.upper(),
|
||
"type": typ,
|
||
})
|
||
except Exception:
|
||
pass
|
||
return results
|
||
|
||
|
||
async def lookup_by_quote(code: str) -> Optional[dict]:
|
||
"""通过行情接口直接查询股票（用于搜索接口不支持的股票）"""
|
||
market = get_market_prefix(code)
|
||
stock_code = f"{market}{code}"
|
||
url = f"https://qt.gtimg.cn/q={stock_code}"
|
||
headers = {"User-Agent": "Mozilla/5.0"}
|
||
async with httpx.AsyncClient() as client:
|
||
try:
|
||
resp = await client.get(url, headers=headers, timeout=10)
|
||
if resp.status_code != 200:
|
||
return None
|
||
text = resp.content.decode("gbk", errors="replace")
|
||
parts = parse_tencent_data(text)
|
||
if not parts or len(parts) < 3:
|
||
return None
|
||
name = parts[1]
|
||
current_price = float(parts[3]) if parts[3] else 0
|
||
if not name or current_price == 0:
|
||
return None
|
||
return {
|
||
"code": code,
|
||
"name": name,
|
||
"market": market.upper(),
|
||
"type": get_board_type(code),
|
||
}
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def get_board_type(code: str) -> str:
|
||
if code.startswith("688"):
|
||
return "GP-A-KCB"
|
||
if code.startswith("300") or code.startswith("301"):
|
||
return "GP-A-CYB"
|
||
if code.startswith("8") or code.startswith("4") or code.startswith("920"):
|
||
return "GP-A-BJB"
|
||
return "GP-A"
|
||
|
||
|
||
async def fetch_quote(code: str) -> Optional[dict]:
|
||
"""获取实时行情"""
|
||
market = get_market_prefix(code)
|
||
stock_code = f"{market}{code}"
|
||
url = f"https://qt.gtimg.cn/q={stock_code}"
|
||
headers = {"User-Agent": "Mozilla/5.0"}
|
||
|
||
async with httpx.AsyncClient() as client:
|
||
try:
|
||
resp = await client.get(url, headers=headers, timeout=10)
|
||
if resp.status_code != 200:
|
||
return None
|
||
text = resp.content.decode("gbk", errors="replace")
|
||
parts = parse_tencent_data(text)
|
||
if not parts or len(parts) < 38:
|
||
return None
|
||
# 字段索引(1-based)：1=名称, 3=当前价, 4=昨收, 5=今开, 6=成交量(手)
|
||
# 7=外盘, 8=内盘, 31=涨跌额, 32=涨跌幅%, 33=最高, 34=最低, 37=成交额(万)
|
||
name = parts[1] or ""
|
||
current_price = float(parts[3]) if parts[3] else 0
|
||
yesterday_close = float(parts[4]) if parts[4] else 0
|
||
today_open = float(parts[5]) if parts[5] else 0
|
||
volume = float(parts[6]) if parts[6] else 0
|
||
high = float(parts[33]) if parts[33] else 0
|
||
low = float(parts[34]) if parts[34] else 0
|
||
amount = float(parts[37]) if parts[37] else 0
|
||
change_val = float(parts[31]) if parts[31] else 0
|
||
change_pct = float(parts[32]) if parts[32] else 0
|
||
outer_disk = float(parts[7]) if parts[7] else 0
|
||
inner_disk = float(parts[8]) if parts[8] else 0
|
||
|
||
if not name or current_price == 0:
|
||
return None
|
||
|
||
change = change_val if change_val != 0 else current_price - yesterday_close
|
||
change_percent = change_pct if change_pct != 0 else (
|
||
(current_price - yesterday_close) / yesterday_close * 100 if yesterday_close > 0 else 0
|
||
)
|
||
|
||
from datetime import datetime
|
||
now = datetime.now()
|
||
return {
|
||
"code": code,
|
||
"market": market.upper(),
|
||
"name": name,
|
||
"todayOpen": today_open,
|
||
"yesterdayClose": yesterday_close,
|
||
"currentPrice": current_price,
|
||
"high": high,
|
||
"low": low,
|
||
"volume": volume * 100,
|
||
"amount": amount,
|
||
"outerDisk": outer_disk,
|
||
"innerDisk": inner_disk,
|
||
"date": now.strftime("%Y-%m-%d"),
|
||
"time": now.strftime("%H:%M:%S"),
|
||
"change": round(change, 2),
|
||
"changePercent": round(change_percent, 2),
|
||
}
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
async def fetch_history(code: str, days: int = 90) -> List[dict]:
|
||
"""获取历史K线（前复权日K），主数据源"""
|
||
market = get_market_prefix(code)
|
||
stock_code = f"{market}{code}"
|
||
url = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={stock_code},day,,,{days},qfq"
|
||
headers = {
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
||
"Referer": "https://stockapp.finance.qq.com/",
|
||
}
|
||
|
||
async with httpx.AsyncClient() as client:
|
||
try:
|
||
resp = await client.get(url, headers=headers, timeout=10)
|
||
if resp.status_code != 200:
|
||
return []
|
||
data = resp.json()
|
||
stock_data = data.get("data", {}).get(stock_code, {})
|
||
raw = stock_data.get("qfqday") or stock_data.get("day") or []
|
||
if not isinstance(raw, list):
|
||
return []
|
||
result = []
|
||
for record in raw:
|
||
if not isinstance(record, list) or len(record) < 6:
|
||
continue
|
||
result.append({
|
||
"date": record[0],
|
||
"open": float(record[1]) if record[1] else 0,
|
||
"close": float(record[2]) if record[2] else 0,
|
||
"high": float(record[3]) if record[3] else 0,
|
||
"low": float(record[4]) if record[4] else 0,
|
||
"volume": int(record[5]) if record[5] else 0,
|
||
})
|
||
return result
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
async def fetch_kline_map(code: str, days: int = 30) -> dict:
|
||
"""获取K线数据并返回 { date: { close, changePercent, turnover } } 映射"""
|
||
market = get_market_prefix(code)
|
||
stock_code = f"{market}{code}"
|
||
url = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={stock_code},day,,,{days + 30},qfq"
|
||
headers = {
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
||
"Referer": "https://stockapp.finance.qq.com/",
|
||
}
|
||
kline_map = {}
|
||
async with httpx.AsyncClient() as client:
|
||
try:
|
||
resp = await client.get(url, headers=headers, timeout=10)
|
||
if resp.status_code != 200:
|
||
return kline_map
|
||
data = resp.json()
|
||
stock_data = data.get("data", {}).get(stock_code, {})
|
||
raw = stock_data.get("qfqday") or stock_data.get("day") or []
|
||
prev_close = 0
|
||
for record in raw:
|
||
if not isinstance(record, list) or len(record) < 6:
|
||
continue
|
||
try:
|
||
date = record[0]
|
||
close = float(record[2]) if record[2] else 0
|
||
turnover = 0
|
||
if len(record) > 6 and isinstance(record[6], (int, float, str)):
|
||
try:
|
||
turnover = float(record[6])
|
||
except (ValueError, TypeError):
|
||
turnover = 0
|
||
change_pct = 0
|
||
if prev_close > 0:
|
||
change_pct = (close - prev_close) / prev_close * 100
|
||
kline_map[date] = {"close": close, "changePercent": change_pct, "turnover": turnover}
|
||
prev_close = close
|
||
except Exception:
|
||
continue
|
||
except Exception:
|
||
pass
|
||
return kline_map
|