dekun/qihuo
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

新增行情K线页，支持分时与多周期图表

Browse Source 
扩展新浪K线拉取与合成逻辑，提供 ECharts 交互图表及实时报价 API。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-15 17:07:04 +08:00
parent 28875078f1
commit 6f3ac3deb6
5 changed files with 621 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kline_chart.py
+179 -21
View File
@@ -24,6 +24,19 @@ PERIOD_MINUTES = {
"4h": "240",
}
MARKET_PERIODS = [
{"key": "timeshare", "label": "分时"},
{"key": "1m", "label": "1分"},
{"key": "2m", "label": "2分"},
{"key": "5m", "label": "5分"},
{"key": "15m", "label": "15分"},
{"key": "1h", "label": "1小时"},
{"key": "2h", "label": "2小时"},
{"key": "4h", "label": "4小时"},
{"key": "d", "label": "日线"},
{"key": "w", "label": "周线"},
]
def ths_to_sina_chart_symbol(symbol: str) -> Optional[str]:
"""ag2608 -> AG2608（新浪 K 线接口合约代码）。"""
@@ -57,15 +70,29 @@ def _parse_jsonp(text: str) -> Optional[list]:
def fetch_sina_klines(symbol: str, period: str) -> list:
"""拉取新浪期货分钟 K 线。"""
"""拉取新浪期货 K 线（原始 bar 列表）。"""
chart_sym = ths_to_sina_chart_symbol(symbol)
if not chart_sym:
return []
if period == "1d":
p = (period or "").lower()
if p in ("1d", "d"):
return _fetch_sina_daily(chart_sym)
typ = PERIOD_MINUTES.get(period)
if not typ:
return []
if p == "w":
return _weekly_from_daily(_fetch_sina_daily(chart_sym))
if p == "timeshare":
bars = _fetch_few_min_line(chart_sym, "1")
return _timeshare_session(bars)
if p == "2m":
return _aggregate_bars(_fetch_few_min_line(chart_sym, "1"), 2)
if p == "2h":
return _aggregate_bars(_fetch_few_min_line(chart_sym, "60"), 2)
typ = PERIOD_MINUTES.get(p)
if typ:
return _fetch_few_min_line(chart_sym, typ)
return []
def _fetch_few_min_line(chart_sym: str, typ: str) -> list:
ts = datetime.now(TZ).strftime("%Y%m%d%H%M%S")
url = (
"https://stock2.finance.sina.com.cn/futures/api/jsonp.php/"
@@ -79,12 +106,131 @@ def fetch_sina_klines(symbol: str, period: str) -> list:
headers={"Referer": "https://finance.sina.com.cn"},
)
bars = _parse_jsonp(resp.text)
return bars or []
return _normalize_bars(bars or [])
except Exception as exc:
logger.warning("fetch kline failed %s %s: %s", chart_sym, period, exc)
logger.warning("fetch kline failed %s %s: %s", chart_sym, typ, exc)
return []
def _normalize_bars(raw: list) -> list:
out = []
for row in raw:
if isinstance(row, list) and len(row) >= 5:
out.append({
"d": str(row[0]),
"o": float(row[1]),
"h": float(row[2]),
"l": float(row[3]),
"c": float(row[4]),
"v": float(row[5]) if len(row) > 5 and row[5] else 0.0,
})
elif isinstance(row, dict) and row.get("d"):
out.append({
"d": str(row["d"]),
"o": float(row.get("o", 0) or 0),
"h": float(row.get("h", 0) or 0),
"l": float(row.get("l", 0) or 0),
"c": float(row.get("c", 0) or 0),
"v": float(row.get("v", 0) or 0),
})
return out
def _aggregate_bars(bars: list, n: int) -> list:
if n <= 1 or not bars:
return bars
out = []
chunk: list = []
for bar in bars:
chunk.append(bar)
if len(chunk) >= n:
out.append(_merge_bars(chunk))
chunk = []
if chunk:
out.append(_merge_bars(chunk))
return out
def _merge_bars(chunk: list) -> dict:
return {
"d": chunk[0]["d"],
"o": chunk[0]["o"],
"h": max(b["h"] for b in chunk),
"l": min(b["l"] for b in chunk),
"c": chunk[-1]["c"],
"v": sum(b.get("v", 0) for b in chunk),
}
def _weekly_from_daily(daily: list) -> list:
if not daily:
return []
buckets: dict[tuple, list] = {}
for bar in daily:
dt = _bar_datetime(bar)
if not dt:
continue
iso = dt.isocalendar()
key = (iso[0], iso[1])
buckets.setdefault(key, []).append(bar)
out = []
for key in sorted(buckets.keys()):
chunk = buckets[key]
out.append(_merge_bars(chunk))
out[-1]["d"] = chunk[-1]["d"]
return out
def _timeshare_session(bars: list) -> list:
if not bars:
return []
today = datetime.now(TZ).date()
session = []
for bar in bars:
dt = _bar_datetime(bar)
if dt and dt.date() == today:
session.append(bar)
if session:
return session[-480:]
return bars[-480:]
def bars_to_api(bars: list) -> list[dict]:
"""转为前端图表 JSON。"""
result = []
for bar in bars:
dt = _bar_datetime(bar)
ts = int(dt.timestamp() * 1000) if dt else None
result.append({
"time": bar["d"],
"timestamp": ts,
"open": bar["o"],
"high": bar["h"],
"low": bar["l"],
"close": bar["c"],
"volume": bar.get("v", 0),
})
return result
def fetch_market_klines(symbol: str, period: str) -> dict:
chart_sym = ths_to_sina_chart_symbol(symbol)
p = (period or "15m").lower()
if p == "timeshare":
chart_type = "line"
else:
chart_type = "candle"
bars = fetch_sina_klines(symbol, p)
return {
"symbol": symbol,
"chart_symbol": chart_sym,
"period": p,
"chart_type": chart_type,
"count": len(bars),
"bars": bars_to_api(bars),
}
def _fetch_sina_daily(chart_sym: str) -> list:
url = (
"https://stock2.finance.sina.com.cn/futures/api/json.php/"
@@ -93,22 +239,34 @@ def _fetch_sina_daily(chart_sym: str) -> list:
try:
resp = requests.get(url, timeout=20, headers={"Referer": "https://finance.sina.com.cn"})
raw = resp.json()
if not raw:
return []
out = []
for row in raw:
if isinstance(row, list) and len(row) >= 5:
out.append({
"d": row[0],
"o": row[1],
"h": row[2],
"l": row[3],
"c": row[4],
})
return out
if raw and isinstance(raw, list):
bars = _normalize_bars(raw)
if bars:
return bars
except Exception as exc:
logger.warning("fetch daily kline failed %s: %s", chart_sym, exc)
return []
return _daily_from_minutes(chart_sym)
def _daily_from_minutes(chart_sym: str) -> list:
"""合约日线接口无数据时，由 60 分钟 K 线按日合成。"""
bars_60 = _fetch_few_min_line(chart_sym, "60")
if not bars_60:
bars_60 = _fetch_few_min_line(chart_sym, "240")
buckets: dict[str, list] = {}
for bar in bars_60:
dt = _bar_datetime(bar)
if not dt:
continue
key = dt.strftime("%Y-%m-%d")
buckets.setdefault(key, []).append(bar)
out = []
for day in sorted(buckets.keys()):
chunk = buckets[day]
merged = _merge_bars(chunk)
merged["d"] = day + " 15:00:00"
out.append(merged)
return out
def _parse_dt(value: str) -> Optional[datetime]: