151 lines
4.4 KiB
Python
151 lines
4.4 KiB
Python
import json
|
|
import urllib.request
|
|
import akshare as ak
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
# Cache for stock master list
|
|
_stock_list_cache: list[dict] = []
|
|
_stock_list_ts: Optional[datetime] = None
|
|
CACHE_TTL = timedelta(hours=6)
|
|
|
|
|
|
def _load_stock_list() -> list[dict]:
|
|
global _stock_list_cache, _stock_list_ts
|
|
now = datetime.now()
|
|
if _stock_list_cache and _stock_list_ts and (now - _stock_list_ts) < CACHE_TTL:
|
|
return _stock_list_cache
|
|
try:
|
|
df = ak.stock_info_a_code_name()
|
|
_stock_list_cache = df.to_dict(orient="records")
|
|
_stock_list_ts = now
|
|
except Exception:
|
|
if _stock_list_cache:
|
|
return _stock_list_cache
|
|
_stock_list_cache = []
|
|
return _stock_list_cache
|
|
|
|
|
|
def search_stocks(query: str, limit: int = 20) -> list[dict]:
|
|
if not query:
|
|
return []
|
|
stocks = _load_stock_list()
|
|
q = query.upper()
|
|
results = []
|
|
for s in stocks:
|
|
code = str(s.get("code", ""))
|
|
name = str(s.get("name", ""))
|
|
if q in code or q in name:
|
|
results.append({"code": code, "name": name})
|
|
if len(results) >= limit:
|
|
break
|
|
return results
|
|
|
|
|
|
def _sina_symbol(code: str) -> str:
|
|
"""Convert 6-digit code to Sina format: sh600519 or sz000001."""
|
|
if code.startswith("6") or code.startswith("9"):
|
|
return f"sh{code}"
|
|
return f"sz{code}"
|
|
|
|
|
|
def _tencent_symbol(code: str) -> str:
|
|
"""Convert to Tencent format."""
|
|
if code.startswith("6") or code.startswith("9"):
|
|
return f"sh{code}"
|
|
return f"sz{code}"
|
|
|
|
|
|
def get_kline(code: str, months: int = 12) -> list[dict]:
|
|
"""Get daily K-line via Sina Finance."""
|
|
end_date = datetime.now()
|
|
start_date = end_date - timedelta(days=months * 31)
|
|
symbol = _sina_symbol(code)
|
|
|
|
# Sina K-line API - scale=240 means daily, datalen estimated
|
|
days_needed = (end_date - start_date).days
|
|
url = (
|
|
f"https://money.finance.sina.com.cn/quotes_service/api/json_v2.php/"
|
|
f"CN_MarketData.getKLineData?symbol={symbol}&scale=240&ma=no&datalen={days_needed}"
|
|
)
|
|
|
|
try:
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)",
|
|
"Referer": "https://finance.sina.com.cn/",
|
|
},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
raw = resp.read().decode("gbk")
|
|
items = json.loads(raw)
|
|
except Exception:
|
|
return []
|
|
|
|
results = []
|
|
for item in items:
|
|
results.append({
|
|
"date": item["day"],
|
|
"open": float(item["open"]),
|
|
"close": float(item["close"]),
|
|
"high": float(item["high"]),
|
|
"low": float(item["low"]),
|
|
"volume": float(item.get("volume", 0)),
|
|
})
|
|
return results
|
|
|
|
|
|
def get_realtime_quote(code: str) -> Optional[dict]:
|
|
"""Get real-time quote via Tencent Finance."""
|
|
symbol = _tencent_symbol(code)
|
|
url = f"https://qt.gtimg.cn/q={symbol}"
|
|
|
|
try:
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)",
|
|
"Referer": "https://gu.qq.com/",
|
|
},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
raw = resp.read().decode("gbk")
|
|
except Exception:
|
|
return None
|
|
|
|
# Tencent format fields:
|
|
# [0]=market [1]=name [2]=code [3]=price [4]=prev_close [5]=open
|
|
# [6]=volume(手) [30]=datetime [31]=change [32]=change_pct [33]=high [34]=low
|
|
try:
|
|
start = raw.index('"') + 1
|
|
end = raw.rindex('"')
|
|
parts = raw[start:end].split("~")
|
|
if len(parts) < 35:
|
|
return None
|
|
|
|
price = float(parts[3])
|
|
prev_close = float(parts[4])
|
|
open_p = float(parts[5])
|
|
high = float(parts[33])
|
|
low = float(parts[34])
|
|
change = float(parts[31])
|
|
change_pct = float(parts[32])
|
|
volume_shares = float(parts[6]) * 100 # 手 -> 股
|
|
|
|
return {
|
|
"code": code,
|
|
"name": parts[1],
|
|
"price": price,
|
|
"change": change,
|
|
"change_pct": change_pct,
|
|
"high": high,
|
|
"low": low,
|
|
"open": open_p,
|
|
"prev_close": prev_close,
|
|
"volume": volume_shares,
|
|
"amount": 0,
|
|
}
|
|
except (ValueError, IndexError):
|
|
return None
|