diff --git a/.codeartsdoer/mem-exp-config.json b/.codeartsdoer/mem-exp-config.json new file mode 100644 index 0000000..f433fd5 --- /dev/null +++ b/.codeartsdoer/mem-exp-config.json @@ -0,0 +1,7 @@ +{ + "auto_experience_retrieve": false, + "enable_experience_collect": false, + "enable_experience_reflect": false, + "scheduled_time_start": "00:00", + "scheduled_time_end": "06:00" +} \ No newline at end of file diff --git a/backend/app/data/market_overview.py b/backend/app/data/market_overview.py index 176f060..1217135 100644 --- a/backend/app/data/market_overview.py +++ b/backend/app/data/market_overview.py @@ -1,6 +1,7 @@ """大盘行情总览（主页展示）——两层结构。 -- live 层：腾讯免费实时行情（qt.gtimg.cn，一次 GET 拿全部指数现价/涨跌幅）， +- live 层：腾讯免费实时行情（qt.gtimg.cn，一次 GET 拿 A股/美股指数现价/涨跌幅）+ + 东财 push2（日韩指数，腾讯 s_ 前缀不覆盖，实测 s_jp*/s_kr* 均无符号）， 进程内缓存 30s（成功）/ 15s 负缓存（失败，防止接口抖动持续拖慢请求）。 - EOD 层：tushare 收盘数据 —— A 股指数 pro.index_daily、全球指数 pro.index_global （45 日 spark 走势）+ 两市统计/成交额历史 pro.daily_info。收盘数据一天一变， @@ -28,7 +29,7 @@ from ..config import settings from .sync_utils import d8_iso, f_clean # (tushare代码, 名称, 地区, 腾讯符号) —— 展示顺序即列表顺序 -# 首页聚焦中美（港股/国际指数在 /indexes 国际指数页）；标普500 腾讯符号是 s_usINX（不是 s_usSPX） +# 首页聚焦中美 + 日韩（其余国际指数在 /indexes 国际指数页）；标普500 腾讯符号是 s_usINX（不是 s_usSPX） MARKET_INDEXES: list[tuple[str, str, str, str]] = [ ("000001.SH", "上证指数", "cn", "s_sh000001"), ("399001.SZ", "深证成指", "cn", "s_sz399001"), @@ -37,20 +38,36 @@ MARKET_INDEXES: list[tuple[str, str, str, str]] = [ ("DJI", "道琼斯", "us", "s_usDJI"), ("IXIC", "纳斯达克", "us", "s_usIXIC"), ("SPX", "标普500", "us", "s_usINX"), + ("N225", "日经225", "apac", ""), # 日韩实时走东财（_APAC_SECIDS），腾讯符号留空 + ("KS11", "韩国KOSPI", "apac", ""), ] # 深证综指：不展示，仅取其 f[7]（深市全市成交额，万元） _TENCENT_SZ_TOTAL = "s_sz399106" -_TENCENT_MAP = {ts_code: sym for ts_code, _, _, sym in MARKET_INDEXES} +_TENCENT_MAP = {ts_code: sym for ts_code, _, _, sym in MARKET_INDEXES if sym} _TENCENT_URL = "http://qt.gtimg.cn/q=" + ",".join([*_TENCENT_MAP.values(), _TENCENT_SZ_TOTAL]) +# 日韩指数实时源：东财 push2 ulist 一次 GET（主站 + 延迟镜像双 host）。腾讯 s_ 前缀不 +# 覆盖日韩（实测均返回 pv_none_match）；新浪 int_kospi 为空且 int_* 行情明显滞后 +# （实测 DJI 差价 ~6000 点），故不用。 +_APAC_SECIDS = {"N225": "100.N225", "KS11": "100.KS11"} +_EM_HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"} # 同 etf_provider +_EM_HOSTS = ["https://push2.eastmoney.com", "https://push2delay.eastmoney.com"] +_EM_PATH = ("/api/qt/ulist.np/get?fltt=2&invt=2&fields=f2,f3,f4,f12,f14&secids=" + + ",".join(_APAC_SECIDS.values())) +# 指数 code -> live 行情键（em: 前缀与腾讯符号隔离），merge 时查表 +_LIVE_SYM_MAP: dict[str, str] = {**_TENCENT_MAP, **{c: f"em:{c}" for c in _APAC_SECIDS}} + _SPARK_DAYS = 45 # 迷你走势取最近 45 个交易日收盘 _HISTORY_DAYS = 150 # 日历日窗口（约 100 个交易日，够取 spark） _CALL_INTERVAL = 0.12 # 顺序调用间隔（秒），对 tushare 控频 -_EOD_KEY = "market_overview:eod:v3" # v3：首页列表移除港股（移入国际指数页），与 v2 隔离 +_EOD_KEY = "market_overview:eod:v4" # v4：新增日韩指数（N225/KS11），与 v3 隔离 _AMOUNT_HIST_CAL_DAYS = 190 # 成交额历史的日历日窗口（≈128 交易日） _AMOUNT_HIST_BARS = 120 # 输出的柱数（取尾部） -_LIVE_FAIL_TTL = 15.0 # live 层失败负缓存（秒） +_LIVE_FAIL_TTL = 15.0 # 腾讯 live 层失败负缓存（秒） +_EM_FAIL_TTL = 90.0 # 东财日韩失败负缓存（秒）：长于 live 缓存 30s， + # 持续故障时最多每 90s 才有一次超时等待，其余刷新直接跳过 class MarketOverviewError(RuntimeError): @@ -69,7 +86,9 @@ def _get_pro(): # ======================= live 层：腾讯实时行情 ======================= _live: dict = {"at": 0.0, "quotes": None} # 进程内缓存（monotonic 时钟） -_live_error: str | None = None # 最近一次实时拉取失败的原因（成功后清空） +_live_error: str | None = None # 最近一次腾讯实时拉取失败的原因（成功后清空） +_live_em_error: str | None = None # 最近一次东财日韩实时拉取失败的原因（成功后清空） +_em_state: dict = {"at": 0.0, "ok": None, "host": 0} # 东财日韩负缓存 + 最近成功 host 索引 async def _fetch_live_http() -> dict[str, dict]: @@ -112,23 +131,91 @@ async def _fetch_live_http() -> dict[str, dict]: return quotes +def _parse_em_ulist(payload: dict) -> dict[str, dict]: + """ulist JSON -> {em:code: {name, price, change, pct, amount_wan}}。fltt=2 下 f2/f3/f4 + 直接是浮点；未开盘等场景字段是 "-"，解析为 None。""" + diff = (payload.get("data") or {}).get("diff") or [] + quotes: dict[str, dict] = {} + for row in diff: + code = str(row.get("f12") or "") + if code not in _APAC_SECIDS: + continue + + def _num(v) -> float | None: + try: + return float(v) + except (TypeError, ValueError): + return None + + quotes[f"em:{code}"] = { + "name": row.get("f14"), + "price": _num(row.get("f2")), + "change": _num(row.get("f4")), + "pct": _num(row.get("f3")), + "amount_wan": None, # 日韩成交额口径不同且未使用，不取 + } + if not quotes: + raise RuntimeError("东财日韩行情响应为空或无法解析") + return quotes + + +async def _fetch_live_apac_http() -> dict[str, dict]: + """日韩指数实时（东财 ulist 一次 GET，主站/延迟镜像按序重试，粘住最近成功的 host）。 + + 坑（同 etf_provider 实测）：push2 主站短连发几次会直接断连，push2delay 镜像稳； + 本机 v2rayN 系统代理(127.0.0.1:10808)对 push2 的 https CONNECT 隧道断连（curl -x + 同样失败，腾讯 http:// 却正常）——必须带浏览器 UA 且 trust_env=False 直连。""" + start = _em_state.get("host", 0) + last_err: Exception | None = None + for i in range(len(_EM_HOSTS)): + url = _EM_HOSTS[(start + i) % len(_EM_HOSTS)] + _EM_PATH + try: + async with httpx.AsyncClient( + timeout=settings.tencent_quote_timeout, headers=_EM_HEADERS, trust_env=False, + ) as client: + resp = await client.get(url) + resp.raise_for_status() + payload = resp.json() + quotes = _parse_em_ulist(payload) + _em_state["host"] = (start + i) % len(_EM_HOSTS) # 粘住成功 host + return quotes + except Exception as e: # noqa: BLE001 —— 换镜像整重来 + last_err = e + raise RuntimeError(f"东财日韩行情双镜像均失败: {str(last_err)[:60]}") + + async def _fetch_live() -> dict[str, dict] | None: - """实时行情（进程内缓存）；任何失败返回 None，上层降级 EOD。""" - global _live_error + """实时行情（进程内缓存）；腾讯与东财各自独立降级，全失败返回 None，上层降级 EOD。 + 东财有独立负缓存（_EM_FAIL_TTL）：持续故障时后续刷新直接跳过，不吃超时等待。""" + global _live_error, _live_em_error now = time.monotonic() age = now - _live["at"] if _live["quotes"] is not None and age < settings.market_live_ttl: return _live["quotes"] if _live["quotes"] is None and _live["at"] > 0 and age < _LIVE_FAIL_TTL: - return None # 负缓存：刚失败过，短时间内不再打腾讯 + return None # 负缓存：刚失败过，短时间内不再打行情接口 + + quotes: dict[str, dict] = {} try: - quotes = await _fetch_live_http() + quotes.update(await _fetch_live_http()) + _live_error = None except Exception as e: # noqa: BLE001 —— 实时层是锦上添花，失败不拖垮整包 - _live.update(at=now, quotes=None) _live_error = f"实时行情: {str(e)[:60]}" + if _em_state["ok"] is False and now - _em_state["at"] < _EM_FAIL_TTL: + pass # 东财刚失败过：负缓存期内跳过（_live_em_error 保留上次原因） + else: + try: + quotes.update(await _fetch_live_apac_http()) + _live_em_error = None + _em_state.update(at=now, ok=True) + except Exception as e: # noqa: BLE001 + _live_em_error = f"日韩实时行情: {str(e)[:60]}" + _em_state.update(at=now, ok=False) + + if not quotes: + _live.update(at=now, quotes=None) return None _live.update(at=now, quotes=quotes) - _live_error = None return quotes @@ -315,7 +402,7 @@ async def fetch_overview(is_trading_day: bool | None = None) -> dict: indexes: list[dict] = [] for it in eod["indexes"]: out = dict(it) - sym = _TENCENT_MAP.get(it["code"]) + sym = _LIVE_SYM_MAP.get(it["code"]) q = live.get(sym) if (live and sym) else None if q and q.get("price") is not None: # spark 永远来自 EOD（末点是上一收盘点，与实时价并存是已知的装饰性差异，不改历史序列） @@ -345,6 +432,8 @@ async def fetch_overview(is_trading_day: bool | None = None) -> dict: errors.append(_eod_refresh_error) if _live_error: errors.append(_live_error) + if _live_em_error: + errors.append(_live_em_error) return { "updated_at": datetime.now().isoformat(), diff --git a/backend/app/screener/market_sync.py b/backend/app/screener/market_sync.py index 21867b8..208b027 100644 --- a/backend/app/screener/market_sync.py +++ b/backend/app/screener/market_sync.py @@ -1,8 +1,9 @@ """全市场数据同步（未复权，写入 candles 全量底座）。 设计：trade_cal 取近 N 个交易日 -> 逐日 pro.daily(trade_date=...) 一次返回全市场当日数据 --> upsert 进 candles（不复权底座，ON CONFLICT 幂等）；daily_basic 仅同步最新交易日到 -DailySnapshot（市值/PE/PB/换手率等截面字段）。 +-> upsert 进 candles（不复权底座，ON CONFLICT 幂等）；daily_basic 同步最新交易日到 +DailySnapshot（市值/PE/PB 等截面字段），并复用该次调用把换手率回写 candles.turnover +（历史缺漏日由自愈循环补，见 _run_sync 第 3.5 步）。 同步为进程内后台任务（MVP 不引入任务队列），前端轮询 /api/screener/sync/status。 daily 与 daily_basic 分步独立落库：daily_basic 积分不足时快照仍可用，错误写入状态不中断任务。 @@ -14,7 +15,7 @@ import logging import time from datetime import datetime, timedelta -from sqlalchemy import delete, func, insert, select +from sqlalchemy import delete, func, insert, select, text from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession @@ -244,7 +245,7 @@ async def _upsert_candle_day(session: AsyncSession, rows: list[dict], listed: se "open": r["open"], "high": r["high"], "low": r["low"], "close": r["close"], "volume": r["vol"] * 100.0, # 手 -> 股 "amount": (r["amount"] * 1000.0) if r["amount"] is not None else None, # 千元 -> 元 - "turnover": None, # 换手率另由 daily_basic 快照维护 + "turnover": None, # 换手率由 _run_sync 第 3/3.5 步从 daily_basic 回写 } for r in rows if plain_code(r["ts_code"]) in listed @@ -269,10 +270,65 @@ async def _upsert_candle_day(session: AsyncSession, rows: list[dict], listed: se await session.commit() -async def _run_sync(days: int, force: bool) -> None: - """后台任务主体：stock_basic -> 逐日日线 -> 最新交易日快照。异常写状态。 +_TURNOVER_FLOOR = "20000104" # daily_basic 最早覆盖日，更早的交易日拉了也是空 - daily_basic 只拉最新交易日（快照条件仅作用于最新截面，且低积分 token 限频 1 次/分钟）。 + +async def _backfill_turnover_day(session: AsyncSession, basic_rows: list[dict], d_str: str) -> int: + """把 daily_basic 的 turnover_rate 回写 candles.turnover（只动该列，幂等）。 + + basic_rows 复用 _fetch_basic 的返回（零额外 API 调用）；单条 UPDATE...FROM + unnest 批量写回，ETF 等不在 daily_basic 的行不会命中。 + """ + syms = [plain_code(r["ts_code"]) for r in basic_rows if r["turnover_rate"] is not None] + trs = [r["turnover_rate"] for r in basic_rows if r["turnover_rate"] is not None] + if not syms: + return 0 + res = await session.execute( + text("UPDATE candles AS c SET turnover = v.t " + "FROM unnest(CAST(:syms AS text[]), CAST(:trs AS float8[])) AS v(sym, t) " + "WHERE c.symbol = v.sym AND c.timeframe = '1d' AND c.ts = :ts"), + {"syms": syms, "trs": trs, "ts": _parse_d(d_str)}, + ) + await session.commit() + return res.rowcount or 0 + + +async def _backfill_turnover_gaps(pro) -> int: + """换手率全范围自愈：按日聚合在市股票的换手覆盖，过半缺失的交易日逐日拉 + daily_basic 补齐，返回处理的缺口天数。 + + 夜间同步与手动同步共用本函数（唯一入口，幂等可断点续跑——补完的日子下轮 + 不再命中），正常无缺口时零 API 调用。只统计在市股票（与 _recent_day_counts + 同口径）：ETF/DEMO 行 daily_basic 天然不覆盖，混进来会把健康日误判成缺换手。 + """ + from ..db import async_session # 延迟导入避免循环 + + async with async_session() as session: + rows = (await session.execute( + select(func.date(Candle.ts), func.count(), func.count(Candle.turnover)) + .where(Candle.timeframe == "1d", Candle.ts >= _parse_d(_TURNOVER_FLOOR), + Candle.symbol.in_(select(StockBasic.symbol).where(StockBasic.list_status == "L"))) + .group_by(func.date(Candle.ts)) + .order_by(func.date(Candle.ts)) + )).all() + gaps = [d.strftime("%Y%m%d") for d, total, done in rows if total and done < total // 2] + for i, d_str in enumerate(gaps, 1): + _sync_state["step"] = f"正在回补 {d_str} 换手率（{i}/{len(gaps)}）" + try: + basic_rows = await asyncio.to_thread(_fetch_basic, pro, d_str) + if basic_rows: + async with async_session() as session: + await _backfill_turnover_day(session, basic_rows, d_str) + except Exception: # noqa: BLE001 —— 单日失败不中断，下次同步再试 + log.warning("换手率回补 %s 失败（下次同步再试）", d_str, exc_info=True) + return len(gaps) + + +async def _run_sync(days: int, force: bool) -> None: + """后台任务主体：stock_basic -> 逐日日线 -> 最新交易日快照 + 换手率回写/自愈。异常写状态。 + + daily_basic 只拉最新交易日（快照条件仅作用于最新截面，且低积分 token 限频 1 次/分钟）； + 历史缺口的换手率由 _backfill_turnover_gaps 统一补齐，夜间/手动同步共用同一管道。 """ from ..db import async_session # 延迟导入避免循环 @@ -343,6 +399,7 @@ async def _run_sync(days: int, force: bool) -> None: async with async_session() as session: latest_dt = await session.scalar(select(func.max(Candle.ts))) latest = latest_dt.strftime("%Y%m%d") if latest_dt else None + basic_rows: list[dict] = [] if latest: async with async_session() as session: have_snap = force or latest not in await _existing_dates(session, DailySnapshot) @@ -353,6 +410,23 @@ async def _run_sync(days: int, force: bool) -> None: async with async_session() as session: await _replace_day(session, DailySnapshot, basic_rows, latest) + # 3.5) 换手率回写：日线同步不写 turnover（daily_basic 才有）——快照那次调用 + # 顺手回写最新日（零额外 API 调用）；历史缺口统一由 _backfill_turnover_gaps + # 全范围扫补，夜间/手动同步共用同一管道 + if latest and basic_rows: + _sync_state["step"] = f"正在回写 {latest} 换手率" + try: + async with async_session() as session: + await _backfill_turnover_day(session, basic_rows, latest) + except Exception: # noqa: BLE001 —— 回写失败不影响快照，缺口由自愈兜底 + log.warning("换手率回写 %s 失败（下次同步自愈）", latest, exc_info=True) + try: + n_gap = await _backfill_turnover_gaps(pro) + if n_gap: + log.info("换手率自愈补齐 %d 个交易日", n_gap) + except Exception: # noqa: BLE001 —— 自愈失败不阻断同步收尾，下次再试 + log.warning("换手率自愈失败（下次同步再试）", exc_info=True) + # candles/复权因子已更新：作废旧 K 线预览缓存（键含版本号，自增即全体失效） await cache.bump_version("candles") # 预热统计缓存：同步任务自己付一次重聚合（>10s）。SWR 下轮询方不等待—— diff --git a/backend/scripts/active_mv_calib.py b/backend/scripts/active_mv_calib.py new file mode 100644 index 0000000..3a9c183 --- /dev/null +++ b/backend/scripts/active_mv_calib.py @@ -0,0 +1,152 @@ +"""0AMV 校验 v2：对照指南针 app EOD 收盘读数，寻找能否精确一致。 + +指南针目标（亿元，app 收盘读数）： + 2026-09-11: 172,364.3 + 2026-09-14: 169,908.5（开 171,979.5 高 175,413.6 低 169,908.5） + 2026-09-15: 166,791.5（开 169,782.7 高 173,060.4 低 166,658.0） + +模型：0AMV = Σ 自由流通市值 × active20，active20 = 1-Π(1-流通换手) 滚动20日。 +自由流通市值 = volume/(turnover_rate_f%) × close（daily_snapshot 近期才有 tff）。 +检验变体：全市场 / 剔北交所 / 再剔科创板 / 剔次新（上市<90自然日）， +若均差一个常数因子 → 指南针自由流通股本口径私有，无法精确复刻。 + +用法（backend 目录）： + env -u SSLKEYLOGFILE PYTHONIOENCODING=utf-8 uv run python scripts/active_mv_calib.py +""" +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import numpy as np +import pandas as pd +from sqlalchemy import text + +from app.db import async_session + +N_WIN = 20 +TARGETS = {"2026-09-11": 172364.3e8, "2026-09-14": 169908.5e8, "2026-09-15": 166791.5e8} + + +async def _load() -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame]: + async with async_session() as session: + hist = (await session.execute(text(""" + SELECT ts, symbol, close, volume, amount, turnover + FROM candles + WHERE timeframe = '1d' + AND turnover IS NOT NULL AND turnover > 0 + AND volume > 0 + AND ts >= now() - interval '500 days' + """))).fetchall() + pend = (await session.execute(text(""" + SELECT ts, symbol, close, volume, amount + FROM candles + WHERE timeframe = '1d' AND volume > 0 + AND ts::date = (SELECT max(ts)::date FROM candles + WHERE timeframe='1d' AND volume > 0) + """))).fetchall() + snap = (await session.execute(text(""" + SELECT ts_code, trade_date::date AS d, turnover_rate, turnover_rate_f + FROM daily_snapshot + WHERE trade_date >= now() - interval '20 days' + AND turnover_rate IS NOT NULL AND turnover_rate_f > 0 + """))).fetchall() + basic = (await session.execute(text(""" + SELECT symbol, market, exchange, list_date FROM stock_basic + """))).fetchall() + return ( + pd.DataFrame(hist, columns=["ts", "symbol", "close", "volume", "amount", "turnover"]), + pd.DataFrame(pend, columns=["ts", "symbol", "close", "volume", "amount"]), + pd.DataFrame(snap, columns=["ts_code", "d", "turnover", "turnover_rate_f"]), + pd.DataFrame(basic, columns=["symbol", "market", "exchange", "list_date"]), + ) + + +def main() -> None: + hist, pend, snap, basic = asyncio.run(_load()) + snap["symbol"] = snap["ts_code"].str.split(".").str[0] + + # 最新 candles 交易日若缺换手率（夜间 3.5 步未跑），用 snapshot 补 + if not pend.empty and not snap.empty: + d_pend = pend["ts"].dt.date.max() + sp = snap[snap["d"] == d_pend][["symbol", "turnover"]] + pend = pend[pend["ts"].dt.date == d_pend].merge(sp, on="symbol", how="inner") + pend = pend[pend["turnover"] > 0] + print(f"最新交易日 {d_pend}：{len(pend):,} 行用 snapshot 换手率补齐") + + df = pd.concat([hist, pend[hist.columns]], ignore_index=True) + df["date"] = df["ts"].dt.date + df = df.sort_values(["symbol", "date"], kind="stable").reset_index(drop=True) + df = df.merge(basic, on="symbol", how="left") + print(f"合计 {len(df):,} 行，{df['symbol'].nunique():,} 只，{df['date'].min()} ~ {df['date'].max()}") + print("板块分布：", df.drop_duplicates("symbol")["market"].value_counts().to_dict()) + + t = (df["turnover"] / 100.0).clip(upper=0.9999) + df["log_inactive"] = np.log1p(-t) + decay = df.groupby("symbol")["log_inactive"].transform( + lambda s: s.rolling(N_WIN, min_periods=1).sum() + ) + df["active20"] = 1.0 - np.exp(decay.to_numpy()) + + # ---- 每个有 tff 的日期：个股级 FF20 明细 ---- + ms: dict = {} + for d in sorted(snap["d"].unique())[-12:]: + day = df[df["date"] == d] + if day.empty: + continue + m = day.merge(snap.loc[snap["d"] == d, ["symbol", "turnover_rate_f"]], + on="symbol", how="inner") + m = m[m["turnover_rate_f"] > 0] + tff = (m["turnover_rate_f"] / 100.0).clip(upper=0.9999) + m["free_mv"] = m["volume"] / tff * m["close"] + m["ff"] = m["free_mv"] * m["active20"] + listed = pd.to_datetime(m["list_date"], format="%Y%m%d", errors="coerce") + m["age_days"] = (pd.Timestamp(d) - listed).dt.days.fillna(10**6) + ms[d] = m + + # ---- 参数扫描：剔除次新窗口 × 是否剔北交所 ---- + print("\n== 参数扫描（对指南针三日的比值；理想=1.00000 稳定）==") + print(f"{'剔北交所':<6}{'剔次新(自然日)':>12}{'09-11':>10}{'09-14':>10}{'09-15':>10}{'均值':>10}{'极差':>9}") + results = [] + for bse_ex in (False, True): + for w in (0, 60, 90, 120, 150, 180, 270, 365, 550): + ratios = [] + for tgt_d, tgt in TARGETS.items(): + d = pd.Timestamp(tgt_d).date() + if d not in ms: + break + m = ms[d] + mask = m["age_days"] >= w + if bse_ex: + mask &= m["market"] != "北交所" + ratios.append(m.loc[mask, "ff"].sum() / tgt) + if len(ratios) < 3: + continue + mean_r = float(np.mean(ratios)) + spread = max(ratios) - min(ratios) + results.append((abs(mean_r - 1) + spread, bse_ex, w, ratios, mean_r, spread)) + print(f"{'是' if bse_ex else '否':<6}{w:>12}{ratios[0]:>10.5f}{ratios[1]:>10.5f}" + f"{ratios[2]:>10.5f}{mean_r:>10.5f}{spread:>9.5f}") + + results.sort() + _, best_bse, best_w, best_ratios, best_mean, best_spread = results[0] + print(f"\n最优配置：剔北交所={'是' if best_bse else '否'}，剔上市<{best_w}自然日") + print(f" 三日比值 {['%.5f' % r for r in best_ratios]}，极差 {best_spread:.5f}") + + # 最优配置下的每日序列 + print("\n== 最优配置近 12 日 FF20（亿）==") + for d in sorted(ms): + m = ms[d] + mask = m["age_days"] >= best_w + if best_bse: + mask &= m["market"] != "北交所" + v = m.loc[mask, "ff"].sum() / 1e8 + mark = f" ←指南针 {TARGETS[str(d)]/1e8:,.1f}" if str(d) in TARGETS else "" + print(f"{d} {v:>10,.0f} n={int(mask.sum())}{mark}") + + +if __name__ == "__main__": + main() diff --git a/backend/scripts/active_mv_probe.py b/backend/scripts/active_mv_probe.py new file mode 100644 index 0000000..3c2b76f --- /dev/null +++ b/backend/scripts/active_mv_probe.py @@ -0,0 +1,116 @@ +"""复刻指南针 0AMV（活跃市值）——可行性验证原型。 + +模型： + 活跃筹码比例(t) = 1 - Π_{近N个交易日} (1 - 换手率) （换手衰减） + 0AMV(t) = Σ_个股 流通市值(t) × 活跃筹码比例(t) +流通市值不额外拉接口：成交量(股) / (换手%/100) 反推流通股本，× close 即得， +且随解禁/增发每日自适应。N 为指南针未公开的窗口参数，输出多组候选供对照 app 校准。 + +用法（backend 目录）： + env -u SSLKEYLOGFILE uv run python scripts/active_mv_probe.py +""" +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import numpy as np +import pandas as pd +from sqlalchemy import text + +from app.db import async_session + +WINDOWS = (20, 60, 120) # 活跃窗口候选（交易日） +LOOKBACK_DAYS = 500 # 日历日回看，足够 120 交易日窗口 +SHOW_DAYS = 20 # 打印最近 N 个交易日 + + +async def _load() -> tuple[pd.DataFrame, pd.DataFrame]: + async with async_session() as session: + rows = (await session.execute(text(""" + SELECT ts, symbol, close, volume, amount, turnover + FROM candles + WHERE timeframe = '1d' + AND turnover IS NOT NULL AND turnover > 0 + AND volume > 0 + AND ts >= now() - interval '500 days' + """))).fetchall() + snap = (await session.execute(text(""" + SELECT ts_code, circ_mv, trade_date + FROM daily_snapshot + WHERE trade_date >= (SELECT max(trade_date) - interval '10 days' FROM daily_snapshot) + AND circ_mv IS NOT NULL + """))).fetchall() + df = pd.DataFrame(rows, columns=["ts", "symbol", "close", "volume", "amount", "turnover"]) + snap = pd.DataFrame(snap, columns=["ts_code", "circ_mv_wan", "snap_date"]) + return df, snap + + +def _validate(df: pd.DataFrame, snap: pd.DataFrame) -> None: + """反推流通市值 vs daily_snapshot（万元→元）交叉验证。""" + if snap.empty: + print("校验：daily_snapshot 为空") + return + snap["symbol"] = snap["ts_code"].str.split(".").str[0] + snap["d"] = snap["snap_date"].dt.date + common = set(snap["d"]) & set(df["date"]) + if not common: + print("校验：daily_snapshot 与 candles 无可对比日期") + return + snap_date = max(common) + day = df[df["date"] == snap_date] + m = day[["symbol", "circ_mv"]].merge( + snap.loc[snap["d"] == snap_date, ["symbol", "circ_mv_wan"]], + on="symbol", how="inner", + ) + if m.empty: + print("校验：无可对比个股") + return + ratio = m["circ_mv"] / (m["circ_mv_wan"] * 1e4) + print(f"校验（{snap_date}，{len(m)} 只）：反推/官方流通市值中位数 = {ratio.median():.4f}，" + f"P10={ratio.quantile(0.1):.3f}，P90={ratio.quantile(0.9):.3f}") + big = m.nlargest(5, "circ_mv") + for _, r in big.iterrows(): + print(f" {r['symbol']}: 反推 {r['circ_mv']/1e8:,.0f}亿 vs 官方 {r['circ_mv_wan']/1e4:,.0f}亿" + f"（比值 {r['circ_mv']/(r['circ_mv_wan']*1e4):.3f}）") + + +def main() -> None: + df, snap = asyncio.run(_load()) + df["date"] = df["ts"].dt.date + # 关键：分组滚动前必须按 (symbol, date) 排序，SQL 返回顺序不保证 + df = df.sort_values(["symbol", "date"], kind="stable").reset_index(drop=True) + print(f"candles 载入 {len(df):,} 行，{df['symbol'].nunique():,} 只，" + f"{df['date'].min()} ~ {df['date'].max()}") + + t = (df["turnover"] / 100.0).clip(upper=0.9999) + df["float_shares"] = df["volume"] / t # 流通股本（股）；volume 单位=股 + df["circ_mv"] = df["float_shares"] * df["close"] # 流通市值（元） + + _validate(df, snap) + + df["log_inactive"] = np.log1p(-t) + agg_cols = {"circ_mv": "sum", "amount": "sum"} + for n in WINDOWS: + decay = df.groupby("symbol")["log_inactive"].transform( + lambda s: s.rolling(n, min_periods=1).sum() + ) + df[f"amv_{n}"] = df["circ_mv"] * (1.0 - np.exp(decay.to_numpy())) + agg_cols[f"amv_{n}"] = "sum" + + daily = df.groupby("date").agg(agg_cols).sort_index().tail(SHOW_DAYS) + print("\n日期 全市场流通市值(万亿) 日成交额(万亿) " + + " ".join(f"0AMV_{n}(万亿) 活跃占比_{n}" for n in WINDOWS)) + for d, r in daily.iterrows(): + print(f"{d} {r['circ_mv']/1e12:8.3f} {r['amount']/1e12:6.3f} " + + " ".join( + f"{r[f'amv_{n}']/1e12:7.3f} {r[f'amv_{n}']/r['circ_mv']*100:5.1f}%" + for n in WINDOWS + )) + + +if __name__ == "__main__": + main() diff --git a/backend/scripts/backfill_turnover.py b/backend/scripts/backfill_turnover.py deleted file mode 100644 index 97c57e1..0000000 --- a/backend/scripts/backfill_turnover.py +++ /dev/null @@ -1,162 +0,0 @@ -"""全量回补换手率（candles.turnover，单位 %）。 - -用法（在 backend 目录下）： - uv run python scripts/backfill_turnover.py # 从 2000-01-01（daily_basic 起点）回补到今天 - uv run python scripts/backfill_turnover.py --start 20200101 - uv run python scripts/backfill_turnover.py --force # 已回补的交易日也重拉 - -- 数据源：Tushare daily_basic(trade_date=..., fields='ts_code,turnover_rate')，按日全市场； -- 幂等可断点续跑：某交易日 candles 已有非空 turnover 即跳过（--force 强制重做）； -- 交易日取自本地 trade_calendar（缓存覆盖不到起点时自动拉一次宽范围日历）； -- 每日一条 UPDATE ... FROM unnest(...) 批量写回，仅更新 turnover 列； -- Tushare 每分钟限频由 _call_retry 自动等待 62s 重试。 - -注意：与 import_tdx_day.py（回填 amount 会整行 upsert）串行运行，避免同表行锁竞争。 -""" -from __future__ import annotations - -import argparse -import asyncio -import sys -import time -from datetime import datetime -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).resolve().parents[1])) - -from app.screener.market_sync import _call_retry, _get_pro - -import asyncpg - - -def load_db_url() -> str: - """与 import_tdx_day.py 相同的 .env -> libpq URL 解析（本地复制避免跨脚本导入）。""" - env = Path(__file__).resolve().parent.parent / ".env" - if env.exists(): - for line in env.read_text(encoding="utf-8").splitlines(): - line = line.strip() - if line.startswith("DATABASE_URL=postgresql+asyncpg://"): - return "postgresql://" + line.split("://", 1)[1] - return "postgresql://postgres:postgres@localhost:5432/stock" - -_DAILY_BASIC_FLOOR = "20000101" # daily_basic 最早覆盖 2000-01-04，更早的交易日无换手数据 -_INTERVAL_MSG = 20 - - -async def _calendar_dates(conn: asyncpg.Connection, start: str, end: str) -> list[str]: - """[start, end] 交易日（升序）。本地缓存覆盖不到起点时拉一次宽范围日历并回写。""" - cached = [r[0] for r in await conn.fetch( - "SELECT trade_date FROM trade_calendar WHERE trade_date >= $1 AND trade_date <= $2 " - "ORDER BY trade_date", start, end)] - if cached and cached[0] <= start: - return cached - - pro = _get_pro() - try: - cal = await asyncio.to_thread( - _call_retry, pro.trade_cal, exchange="SSE", start_date=start, end_date=end, is_open="1" - ) - dates = sorted(cal["cal_date"].tolist()) - except Exception as e: # noqa: BLE001 - if not cached: - raise - print(f"交易日历拉取受限（{str(e)[:100]}），沿用本地缓存") - return cached - have = set(cached) - fresh = [d for d in dates if d not in have] - if fresh: - await conn.executemany( - "INSERT INTO trade_calendar (trade_date) VALUES ($1) ON CONFLICT DO NOTHING", [(d,) for d in fresh] - ) - return dates - - -async def _day_status(conn: asyncpg.Connection, d: str) -> tuple[int, int]: - """(已有换手的行数, 当日总行数)。无行情的日子 total=0 直接跳过。""" - row = await conn.fetchrow( - "SELECT count(*) FILTER (WHERE turnover IS NOT NULL) AS done, count(*) AS total " - "FROM candles WHERE timeframe = '1d' AND ts = $1::timestamp", datetime.strptime(d, "%Y%m%d") - ) - return row["done"], row["total"] - - -async def main(start: str, end: str, force: bool) -> None: - conn = await asyncpg.connect(load_db_url()) - try: - # 默认起点：daily_basic 覆盖范围与 candles 最早日线的较大者（更早的日期拉了也是空） - if start is None: - first = await conn.fetchval( - "SELECT min(ts) FROM candles WHERE timeframe = '1d' AND symbol <> 'DEMO'") - start = max(first.strftime("%Y%m%d"), _DAILY_BASIC_FLOOR) if first else _DAILY_BASIC_FLOOR - if end is None: - end = datetime.now().strftime("%Y%m%d") - - dates = await _calendar_dates(conn, start, end) - todo: list[str] = [] - for d in dates: - if force: - done, total = await _day_status(conn, d) - if total: - todo.append(d) - continue - done, total = await _day_status(conn, d) - if total and done < total // 2: # 过半缺换手才重做（容忍个别股票无快照） - todo.append(d) - print(f"区间 {start}~{end} 共 {len(dates)} 个交易日，待回补 {len(todo)} 个") - - pro = _get_pro() - done = 0 - t0 = time.time() - for d in todo: - time.sleep(0.15) # 轻微控频；分钟级限频由 _call_retry 自动等待重试 - df = None - for attempt in range(5): # 网络抖动（超时/断连）也重试，_call_retry 只兜限频 - try: - df = _call_retry( - pro.daily_basic, trade_date=d, fields="ts_code,trade_date,turnover_rate" - ) - break - except Exception as e: # noqa: BLE001 - wait = min(30 * (attempt + 1), 120) - print(f" {d} 拉取异常（{str(e)[:80]}），{wait}s 后重试 {attempt + 1}/5") - time.sleep(wait) - if df is None: - print(f" {d} 连续 5 次失败，跳过（断点续跑可补）") - continue - if df.empty: - continue - - syms: list[str] = [] - vals: list[float] = [] - for _, r in df.iterrows(): - tr = r["turnover_rate"] - if tr is None or tr != tr: # None / NaN - continue - syms.append(str(r["ts_code"]).split(".")[0]) - vals.append(float(tr)) - if not syms: - continue - n = await conn.execute( - "UPDATE candles AS c SET turnover = v.t " - "FROM unnest($1::text[], $2::float8[]) AS v(sym, t) " - "WHERE c.symbol = v.sym AND c.timeframe = '1d' AND c.ts = $3::timestamp", - syms, vals, datetime.strptime(d, "%Y%m%d"), - ) - done += 1 - if done % _INTERVAL_MSG == 0 or done == len(todo): - elapsed = time.time() - t0 - eta = elapsed / done * (len(todo) - done) if done else 0 - print(f" 进度 {done}/{len(todo)}（{d}，{len(syms)} 只，{n}），" - f"{elapsed:.0f}s 已用，预计还需 {eta/60:.0f}m") - print(f"回补完成：{done} 个交易日") - finally: - await conn.close() - - -if __name__ == "__main__": - ap = argparse.ArgumentParser(description="全量回补换手率 candles.turnover") - ap.add_argument("--start", default=None, help="YYYYMMDD，默认 max(candles 最早, 20000101)") - ap.add_argument("--end", default=None, help="YYYYMMDD，默认今天") - ap.add_argument("--force", action="store_true", help="已有换手的交易日也重拉") - a = ap.parse_args() - asyncio.run(main(a.start, a.end, a.force)) diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index 1d3a9a3..b287ac6 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -435,7 +435,7 @@ export interface EventBacktestResponse { export interface IndexQuote { code: string; // 000001.SH / HKTECH / DJI name: string; // 上证指数 / 恒生科技 / 道琼斯 - region: 'cn' | 'hk' | 'us'; + region: 'cn' | 'hk' | 'us' | 'apac'; close: number | null; change: number | null; pct_chg: number | null; diff --git a/frontend/src/components/MarketOverview.vue b/frontend/src/components/MarketOverview.vue index 1ca71ab..c9b5746 100644 --- a/frontend/src/components/MarketOverview.vue +++ b/frontend/src/components/MarketOverview.vue @@ -8,8 +8,9 @@ import type { MarketOverview as Overview } from '@/api/types'; import AmountHistoryChart from '@/components/AmountHistoryChart.vue'; import Sparkline from '@/components/Sparkline.vue'; -// 主页大盘总览：中美指数最近收盘（收盘口径，标注交易日），沪深两市市值/成交统计。 -// 港股与国际指数在 /indexes 国际指数页。数据为 EOD 口径，进页面拉一次 + 手动刷新即可。 +// 主页大盘总览：沪深 + 海外（美/日/韩）指数最近收盘（收盘口径，标注交易日），两市市值/成交统计。 +// 其余国际指数在 /indexes 国际指数页。数据为 EOD 口径，进页面拉一次 + 手动刷新即可。 +// 卡片整体可点击，跳 /indexes/:code 指数详情（K 线 / 基本信息 / 估值）。 const overview = ref(null); const loading = ref(false); @@ -32,10 +33,21 @@ const groups = computed(() => { const idx = overview.value?.indexes ?? []; return [ { label: '沪深主要指数', items: idx.filter((i) => i.region === 'cn') }, - { label: '美股市场', items: idx.filter((i) => i.region === 'us') }, + // 美/日/韩合并一行，避免首页纵向空间被三组卡片撑爆 + { label: '海外市场', items: idx.filter((i) => i.region === 'us' || i.region === 'apac') }, ].filter((g) => g.items.length > 0); }); +/** 每组卡片列数：4 列（沪深）/ 5 列（海外，md 起 3 列防挤） */ +function gridClass(n: number): string { + if (n === 4) return 'md:grid-cols-4'; + return 'md:grid-cols-3 lg:grid-cols-5'; +} + +function cardTo(code: string): string { + return `/indexes/${encodeURIComponent(code)}`; +} + const updatedAt = computed(() => { const s = overview.value?.updated_at; return s ? new Date(s).toLocaleTimeString('zh-CN', { hour: '2-digit', minute: '2-digit' }) : ''; @@ -107,10 +119,10 @@ function fmtYi(v: number | null | undefined): string {
-
+
-
-
+
+
@@ -123,14 +135,13 @@ function fmtYi(v: number | null | undefined): string {