Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
abd4d22a67 | ||
|
|
f014eb11bd |
+240
-1351
File diff suppressed because it is too large
Load Diff
@@ -1,18 +1,16 @@
|
|||||||
<!DOCTYPE html>
|
<!DOCTYPE html>
|
||||||
<html lang="zh-CN">
|
<html lang="zh-CN">
|
||||||
<head>
|
<head>
|
||||||
<meta charset="UTF-8" />
|
<meta charset="UTF-8" />
|
||||||
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
|
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
||||||
<title>小白复盘 · 数据中枢</title>
|
<title>xiaobai-datahub 管理后台</title>
|
||||||
<link rel="stylesheet" href="/admin/styles.css" />
|
<link rel="stylesheet" href="/admin/styles.css" />
|
||||||
</head>
|
</head>
|
||||||
<body>
|
<body>
|
||||||
<div id="app">
|
<div id="app">
|
||||||
|
<section id="login-view" class="panel auth-panel">
|
||||||
<div id="login-view" class="auth-wrap bg-ambient" hidden>
|
|
||||||
<section class="panel auth-panel">
|
|
||||||
<h1>数据中枢</h1>
|
<h1>数据中枢</h1>
|
||||||
<p class="muted">内网管理后台 · 运行总览 / 数据源配置 / 数据血缘。</p>
|
<p class="muted">内网管理后台，用于查看源状态、调度和盘后发布批次。</p>
|
||||||
<form id="login-form">
|
<form id="login-form">
|
||||||
<label>账号 <input name="username" value="hub_admin" autocomplete="username" /></label>
|
<label>账号 <input name="username" value="hub_admin" autocomplete="username" /></label>
|
||||||
<label>密码 <input name="password" type="password" autocomplete="current-password" /></label>
|
<label>密码 <input name="password" type="password" autocomplete="current-password" /></label>
|
||||||
@@ -20,95 +18,36 @@
|
|||||||
<p id="login-error" class="error" hidden></p>
|
<p id="login-error" class="error" hidden></p>
|
||||||
</form>
|
</form>
|
||||||
</section>
|
</section>
|
||||||
</div>
|
|
||||||
|
|
||||||
<div id="change-view" class="auth-wrap bg-ambient" hidden>
|
<section id="change-view" class="panel auth-panel" hidden>
|
||||||
<section class="panel auth-panel">
|
|
||||||
<h1>修改初始密码</h1>
|
<h1>修改初始密码</h1>
|
||||||
<p class="muted">首次登录需先设置新密码才能进入。</p>
|
|
||||||
<form id="change-form">
|
<form id="change-form">
|
||||||
<label>当前密码 <input name="current" type="password" autocomplete="current-password" /></label>
|
<label>当前密码 <input name="current" type="password" /></label>
|
||||||
<label>新密码（至少 8 位） <input name="new_password" type="password" autocomplete="new-password" /></label>
|
<label>新密码（至少 8 位） <input name="new_password" type="password" /></label>
|
||||||
<button type="submit">保存并继续</button>
|
<button type="submit">保存并继续</button>
|
||||||
<p id="change-error" class="error" hidden></p>
|
<p id="change-error" class="error" hidden></p>
|
||||||
</form>
|
</form>
|
||||||
</section>
|
</section>
|
||||||
|
|
||||||
|
<section id="shell" hidden>
|
||||||
|
<header class="top">
|
||||||
|
<strong>xiaobai-datahub</strong>
|
||||||
|
<span id="phase" class="pill"></span>
|
||||||
|
<span id="who" class="muted"></span>
|
||||||
|
<button type="button" id="theme-btn" class="ghost">夜间</button>
|
||||||
|
<button type="button" id="logout-btn" class="ghost">退出</button>
|
||||||
|
</header>
|
||||||
|
<nav>
|
||||||
|
<button data-page="overview" class="active">总览</button>
|
||||||
|
<button data-page="sources">数据源</button>
|
||||||
|
<button data-page="jobs">调度任务</button>
|
||||||
|
<button data-page="release">盘后发布</button>
|
||||||
|
<button data-page="datasets">数据集</button>
|
||||||
|
<button data-page="audit">审计</button>
|
||||||
|
</nav>
|
||||||
|
<main id="page"></main>
|
||||||
|
</section>
|
||||||
</div>
|
</div>
|
||||||
|
<script src="/admin/app.js"></script>
|
||||||
<div class="app bg-ambient" id="appRoot" hidden>
|
|
||||||
<div class="bg-layer bg-gridlines z0"></div>
|
|
||||||
<div class="bg-layer scanlines z50"></div>
|
|
||||||
|
|
||||||
<header class="hdr">
|
|
||||||
<div class="hdr-in">
|
|
||||||
<div class="radar" style="width:26px;height:26px" id="radarLogo">
|
|
||||||
<div class="sweep radar-sweep"></div>
|
|
||||||
<div class="ring1" style="inset:5.72px"></div>
|
|
||||||
<div class="ring2" style="inset:9.88px"></div>
|
|
||||||
<div class="center"></div>
|
|
||||||
<div class="blip pulse-dot" id="radarBlipOk" style="left:64%;top:30%;background:#34d399"></div>
|
|
||||||
<div class="blip pulse-dot" id="radarBlipBad" style="left:30%;top:62%;background:#f87171;animation-delay:.6s;display:none"></div>
|
|
||||||
</div>
|
|
||||||
<div class="brand"><span class="b1">小白复盘 <span style="color:#22d3ee">·</span> 数据中枢</span><span class="b2">DATA-HUB</span></div>
|
|
||||||
<span class="vsep"></span>
|
|
||||||
<nav class="nav" id="navEl">
|
|
||||||
<button class="navbtn" data-nav="overview"><span class="tri">▸</span>运行总览<span class="en">OVERVIEW</span></button>
|
|
||||||
<button class="navbtn" data-nav="sources"><span class="tri">▸</span>数据源配置<span class="en">SOURCES</span></button>
|
|
||||||
<button class="navbtn" data-nav="lineage"><span class="tri">▸</span>数据血缘<span class="en">LINEAGE</span></button>
|
|
||||||
</nav>
|
|
||||||
<span class="flex1"></span>
|
|
||||||
<span class="mdtag" id="phaseTag"></span>
|
|
||||||
<span class="livespan" id="liveSpan"><span class="livedot pulse-dot"></span>LIVE</span>
|
|
||||||
<span class="clock num" id="clock"><span id="ckD"></span><span class="csep">|</span><span class="ct"><span id="ckH"></span><span class="blink cc">:</span><span id="ckM"></span><span class="blink cc">:</span><span id="ckS"></span></span></span>
|
|
||||||
<span class="vsep"></span>
|
|
||||||
<button class="tbtn" id="opsBtn" style="font-size:10px">调度 / 发布 / 审计</button>
|
|
||||||
<button class="tbtn" id="calmBtn" style="font-size:10px">减少动态</button>
|
|
||||||
<span id="who" class="muted" style="font-size:10px"></span>
|
|
||||||
<button class="tbtn" id="logout-btn" style="font-size:10px">退出</button>
|
|
||||||
</div>
|
|
||||||
</header>
|
|
||||||
|
|
||||||
<main class="main" id="mainEl"></main>
|
|
||||||
|
|
||||||
<footer class="tape">
|
|
||||||
<span class="tape-label" id="tapeLed"><span class="led rev pulse"></span>EVENT TAPE</span>
|
|
||||||
<div class="tape-view"><div class="marquee-track" id="marqueeTrack"><span style="display:inline-flex;align-items:center" id="tapeA"></span><span style="display:inline-flex;align-items:center" id="tapeB"></span></div></div>
|
|
||||||
</footer>
|
|
||||||
</div>
|
|
||||||
|
|
||||||
<div class="drawer-mask" id="drawerMask">
|
|
||||||
<aside class="drawer">
|
|
||||||
<div class="drawer-hd">
|
|
||||||
<span class="ttl">调度 / 发布 / 审计</span>
|
|
||||||
<span class="sub" id="drawerSub"></span>
|
|
||||||
<button class="drawer-close" id="drawerClose">×</button>
|
|
||||||
</div>
|
|
||||||
<div class="drawer-tabs" id="drawerTabs">
|
|
||||||
<button class="drawer-tab" data-dtab="jobs">调度任务</button>
|
|
||||||
<button class="drawer-tab" data-dtab="release">盘后发布</button>
|
|
||||||
<button class="drawer-tab" data-dtab="audit">审计</button>
|
|
||||||
</div>
|
|
||||||
<div class="drawer-body" id="drawerBody"></div>
|
|
||||||
</aside>
|
|
||||||
</div>
|
|
||||||
|
|
||||||
<div class="modal-mask" id="modalMask">
|
|
||||||
<div class="modal-box">
|
|
||||||
<h3 id="modalTitle">危险操作确认</h3>
|
|
||||||
<p id="modalDesc"></p>
|
|
||||||
<label>管理密码 <input id="modalPassword" type="password" autocomplete="current-password" /></label>
|
|
||||||
<label id="modalConfirmWrap">请输入确认词 <span id="modalConfirmWord" class="num" style="color:#fbbf24"></span> <input id="modalConfirm" type="text" autocomplete="off" /></label>
|
|
||||||
<p class="modal-err" id="modalErr"></p>
|
|
||||||
<div class="modal-actions">
|
|
||||||
<button class="tbtn" id="modalCancel">取消</button>
|
|
||||||
<button class="tbtn danger" id="modalOk">确认执行</button>
|
|
||||||
</div>
|
|
||||||
</div>
|
|
||||||
</div>
|
|
||||||
|
|
||||||
<div class="toast" id="toastEl"></div>
|
|
||||||
|
|
||||||
</div>
|
|
||||||
<script src="/admin/app.js"></script>
|
|
||||||
</body>
|
</body>
|
||||||
</html>
|
</html>
|
||||||
|
|||||||
File diff suppressed because one or more lines are too long
@@ -31,18 +31,10 @@ class AdminAPI:
|
|||||||
)
|
)
|
||||||
is_open = bool(cal and int(cal["is_open"]) == 1)
|
is_open = bool(cal and int(cal["is_open"]) == 1)
|
||||||
pubs = self.db.fetchall("SELECT * FROM publications WHERE trade_date = ?", (today,))
|
pubs = self.db.fetchall("SELECT * FROM publications WHERE trade_date = ?", (today,))
|
||||||
# HEL-529 fix 2: "待处理"只保留当前仍未恢复的最新异常。一条失败/停滞
|
failed = self.db.fetchall(
|
||||||
# 批次若已被同数据集更晚的成功批次或当日发布解决，就不再是当前故障，
|
"SELECT * FROM batches WHERE trade_date = ? AND state IN ('failed','staged')",
|
||||||
# 历史记录仍完整保留在 batches/audit 明细里，不在此重复展示。
|
(today,),
|
||||||
latest_by_dataset: dict[str, dict[str, Any]] = {}
|
)
|
||||||
for row in self.db.fetchall("SELECT * FROM batches WHERE trade_date = ? ORDER BY started_at, batch_id", (today,)):
|
|
||||||
latest_by_dataset[row["dataset"]] = row
|
|
||||||
published_datasets = {row["dataset"] for row in pubs if row["state"] == "published"}
|
|
||||||
anomalies = [
|
|
||||||
row
|
|
||||||
for dataset, row in latest_by_dataset.items()
|
|
||||||
if dataset not in published_datasets and row["state"] in ("failed", "staged")
|
|
||||||
]
|
|
||||||
calls = self.db.fetchall(
|
calls = self.db.fetchall(
|
||||||
"SELECT * FROM src_calls ORDER BY id DESC LIMIT 20",
|
"SELECT * FROM src_calls ORDER BY id DESC LIMIT 20",
|
||||||
)
|
)
|
||||||
@@ -53,7 +45,7 @@ class AdminAPI:
|
|||||||
"eod_status": self.scheduler.eod_status(today),
|
"eod_status": self.scheduler.eod_status(today),
|
||||||
"revision_status": self.scheduler.revision_status(today),
|
"revision_status": self.scheduler.revision_status(today),
|
||||||
"publications": pubs,
|
"publications": pubs,
|
||||||
"anomalies": anomalies,
|
"anomalies": failed,
|
||||||
"recent_calls": _public_calls(calls),
|
"recent_calls": _public_calls(calls),
|
||||||
"source_count": len(self.db.fetchall("SELECT provider FROM src_health")),
|
"source_count": len(self.db.fetchall("SELECT provider FROM src_health")),
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -38,7 +38,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "calendar",
|
"dataset": "calendar",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "每日 08:45 预检",
|
|
||||||
"v1_endpoint": "/v1/calendar",
|
"v1_endpoint": "/v1/calendar",
|
||||||
"primary_source": "tushare:trade_cal",
|
"primary_source": "tushare:trade_cal",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -47,7 +46,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "stocks",
|
"dataset": "stocks",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "每日 20:00 / 23:10",
|
|
||||||
"v1_endpoint": "/v1/stocks",
|
"v1_endpoint": "/v1/stocks",
|
||||||
"primary_source": "tushare:stock_basic",
|
"primary_source": "tushare:stock_basic",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -56,7 +54,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "daily",
|
"dataset": "daily",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "盘后 15:05 · 重试至 23:30",
|
|
||||||
"v1_endpoint": "/v1/bars/daily",
|
"v1_endpoint": "/v1/bars/daily",
|
||||||
"primary_source": "tushare:daily",
|
"primary_source": "tushare:daily",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -65,7 +62,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "valuation",
|
"dataset": "valuation",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "盘后 15:05 · 复核 20:00",
|
|
||||||
"v1_endpoint": "/v1/valuation",
|
"v1_endpoint": "/v1/valuation",
|
||||||
"primary_source": "tushare:daily_basic",
|
"primary_source": "tushare:daily_basic",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -74,7 +70,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "moneyflow",
|
"dataset": "moneyflow",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "盘后 15:05 · 重试至 23:30",
|
|
||||||
"v1_endpoint": "/v1/moneyflow",
|
"v1_endpoint": "/v1/moneyflow",
|
||||||
"primary_source": "tushare:moneyflow",
|
"primary_source": "tushare:moneyflow",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -83,7 +78,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "auction",
|
"dataset": "auction",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "盘后 15:05",
|
|
||||||
"v1_endpoint": "/v1/auction",
|
"v1_endpoint": "/v1/auction",
|
||||||
"primary_source": "tushare:stk_auction",
|
"primary_source": "tushare:stk_auction",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -92,7 +86,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "index_daily",
|
"dataset": "index_daily",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "盘后 15:10 · 重试至 23:30",
|
|
||||||
"v1_endpoint": "/v1/indexes/bars",
|
"v1_endpoint": "/v1/indexes/bars",
|
||||||
"primary_source": "tushare:index_daily",
|
"primary_source": "tushare:index_daily",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -101,7 +94,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "limit_events",
|
"dataset": "limit_events",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "盘后 16:40 · 重试至 23:30",
|
|
||||||
"v1_endpoint": "/v1/limit-events",
|
"v1_endpoint": "/v1/limit-events",
|
||||||
"primary_source": "tushare:limit_list_d",
|
"primary_source": "tushare:limit_list_d",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -110,7 +102,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "popularity",
|
"dataset": "popularity",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "盘后 22:40",
|
|
||||||
"v1_endpoint": "/v1/popularity",
|
"v1_endpoint": "/v1/popularity",
|
||||||
"primary_source": "tushare:ths_hot+dc_hot",
|
"primary_source": "tushare:ths_hot+dc_hot",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -119,7 +110,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "dragon_tiger",
|
"dataset": "dragon_tiger",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "盘后 16:45 · 重试至 23:30",
|
|
||||||
"v1_endpoint": "/v1/dragon-tiger",
|
"v1_endpoint": "/v1/dragon-tiger",
|
||||||
"primary_source": "tushare:hm_detail",
|
"primary_source": "tushare:hm_detail",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -128,7 +118,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "sector_daily",
|
"dataset": "sector_daily",
|
||||||
"tier": "official",
|
"tier": "official",
|
||||||
"update_freq": "盘后 15:20 · 重试至 23:30",
|
|
||||||
"v1_endpoint": "/v1/sectors",
|
"v1_endpoint": "/v1/sectors",
|
||||||
"primary_source": "tushare:ths_daily+dc_index+sw_daily",
|
"primary_source": "tushare:ths_daily+dc_index+sw_daily",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -137,7 +126,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "quotes_latest",
|
"dataset": "quotes_latest",
|
||||||
"tier": "provisional",
|
"tier": "provisional",
|
||||||
"update_freq": "盘中 · 缓存 60s",
|
|
||||||
"v1_endpoint": "/v1/quotes/latest",
|
"v1_endpoint": "/v1/quotes/latest",
|
||||||
"primary_source": "eastmoney:ulist/clist",
|
"primary_source": "eastmoney:ulist/clist",
|
||||||
"backup_source": "tencent:qt",
|
"backup_source": "tencent:qt",
|
||||||
@@ -146,7 +134,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "index_quotes",
|
"dataset": "index_quotes",
|
||||||
"tier": "provisional",
|
"tier": "provisional",
|
||||||
"update_freq": "盘中 · 缓存 60s",
|
|
||||||
"v1_endpoint": "/v1/indexes/quotes",
|
"v1_endpoint": "/v1/indexes/quotes",
|
||||||
"primary_source": "eastmoney:ulist",
|
"primary_source": "eastmoney:ulist",
|
||||||
"backup_source": "tencent:qt",
|
"backup_source": "tencent:qt",
|
||||||
@@ -155,7 +142,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "sectors_quote",
|
"dataset": "sectors_quote",
|
||||||
"tier": "provisional",
|
"tier": "provisional",
|
||||||
"update_freq": "盘中 · 缓存 60s",
|
|
||||||
"v1_endpoint": "/v1/sectors/quote",
|
"v1_endpoint": "/v1/sectors/quote",
|
||||||
"primary_source": "eastmoney:sw",
|
"primary_source": "eastmoney:sw",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -164,7 +150,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "limit_pool",
|
"dataset": "limit_pool",
|
||||||
"tier": "provisional",
|
"tier": "provisional",
|
||||||
"update_freq": "盘中 · 缓存 60s",
|
|
||||||
"v1_endpoint": "/v1/limit-pool",
|
"v1_endpoint": "/v1/limit-pool",
|
||||||
"primary_source": "eastmoney:zt_pool",
|
"primary_source": "eastmoney:zt_pool",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -173,7 +158,6 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "intraday_points",
|
"dataset": "intraday_points",
|
||||||
"tier": "provisional",
|
"tier": "provisional",
|
||||||
"update_freq": "盘中 · 缓存 20s",
|
|
||||||
"v1_endpoint": "/v1/intraday/points",
|
"v1_endpoint": "/v1/intraday/points",
|
||||||
"primary_source": "eastmoney:trends2",
|
"primary_source": "eastmoney:trends2",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
@@ -182,33 +166,10 @@ DATASETS: list[dict[str, Any]] = [
|
|||||||
{
|
{
|
||||||
"dataset": "ifind_wencai",
|
"dataset": "ifind_wencai",
|
||||||
"tier": "licensed",
|
"tier": "licensed",
|
||||||
"update_freq": "按需调用",
|
|
||||||
"v1_endpoint": "/v1/query (api_name=ifind_wencai)",
|
"v1_endpoint": "/v1/query (api_name=ifind_wencai)",
|
||||||
"primary_source": "ifind:wencai",
|
"primary_source": "ifind:smart_stock_picking",
|
||||||
"backup_source": None,
|
"backup_source": None,
|
||||||
# Real call site: backend/features/pools/service.py:82 (wencai(query,"stock"))
|
"known_consumers": ["问师(自然语言选股，需 iFinD 凭证)"],
|
||||||
# feeds 股票池's ifind_event_enrichment_v1 (涨停/炸板/跌停原因、首末
|
|
||||||
# 涨停时间、开板次数). 问师 does NOT call wencai anywhere — its main
|
|
||||||
# body is LLM + 主站市场快照 (mentor/service.py builds context from
|
|
||||||
# dashboard/popularity snapshots only).
|
|
||||||
"known_consumers": ["股票池(涨停/炸板/跌停事件补充 enrichment，需 iFinD 凭证)"],
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"dataset": "ifind_history",
|
|
||||||
"tier": "licensed",
|
|
||||||
"update_freq": "问师问询时按需 · 45 日回看",
|
|
||||||
"v1_endpoint": "/v1/query (api_name=ifind_history)",
|
|
||||||
"primary_source": "ifind:history",
|
|
||||||
"backup_source": None,
|
|
||||||
# Real call site: backend/features/mentor/service.py:433-457
|
|
||||||
# (_mentor_market_matrix → ifind.history(close/volume/amount, 45 日回看)).
|
|
||||||
# Only the trend/macro thinking-model profiles use it, and only when
|
|
||||||
# iFinD is configured — it fails open to [] otherwise. 问师其余子能力
|
|
||||||
# (本体问答/低吸/人气上下文等) 不依赖 iFinD.
|
|
||||||
"known_consumers": [
|
|
||||||
"问师·趋势思维模型(指数动量矩阵，可选)",
|
|
||||||
"问师·宏观思维模型(宽基指数与核心ETF矩阵，可选)",
|
|
||||||
],
|
|
||||||
},
|
},
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|||||||
@@ -99,58 +99,6 @@ STAGING_INSERT = {
|
|||||||
**EXTENDED_STAGING_INSERT,
|
**EXTENDED_STAGING_INSERT,
|
||||||
}
|
}
|
||||||
|
|
||||||
# Staging-table business keys (PRIMARY KEY minus the constant batch_id),
|
|
||||||
# mirroring the PRIMARY KEY clauses declared in datahub/db.py and
|
|
||||||
# datahub/datasets_ext.py. Used only to collapse within-batch duplicates so
|
|
||||||
# a single upstream response cannot fail the whole batch on a UNIQUE
|
|
||||||
# constraint (HEL-529: 2026-09-14 dc_hot returned 4 duplicate ts_codes and
|
|
||||||
# hm_detail 43 duplicate (ts_code, hm_name) keys in one response, which has
|
|
||||||
# blocked popularity/dragon_tiger publishing every day since 09-07).
|
|
||||||
STAGING_KEY_FIELDS = {
|
|
||||||
"stocks": ("ts_code", "trade_date"),
|
|
||||||
"daily": ("ts_code", "trade_date"),
|
|
||||||
"valuation": ("ts_code", "trade_date"),
|
|
||||||
"moneyflow": ("ts_code", "trade_date"),
|
|
||||||
"auction": ("ts_code", "trade_date"),
|
|
||||||
"index_daily": ("ts_code", "trade_date"),
|
|
||||||
"limit_events": ("ts_code", "trade_date", "limit_type"),
|
|
||||||
"popularity": ("ts_code", "trade_date", "source"),
|
|
||||||
"dragon_tiger": ("ts_code", "trade_date", "hm_name"),
|
|
||||||
"sector_daily": ("ts_code", "trade_date", "family"),
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def _dedupe_staging_rows(dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
||||||
"""Collapse within-batch duplicates on the staging table's business key.
|
|
||||||
|
|
||||||
Deterministic: keeps the LAST occurrence of each key (the same row the
|
|
||||||
eod copy's INSERT OR REPLACE would keep), preserves first-seen order, and
|
|
||||||
never touches rows across batches. Datasets without a declared business
|
|
||||||
key are returned unchanged.
|
|
||||||
"""
|
|
||||||
fields = STAGING_KEY_FIELDS.get(dataset)
|
|
||||||
if not fields:
|
|
||||||
return rows
|
|
||||||
seen: dict[tuple, int] = {}
|
|
||||||
out: list[dict[str, Any]] = []
|
|
||||||
dropped = 0
|
|
||||||
for row in rows:
|
|
||||||
key = tuple(row.get(field) for field in fields)
|
|
||||||
if key in seen:
|
|
||||||
out[seen[key]] = row
|
|
||||||
dropped += 1
|
|
||||||
else:
|
|
||||||
seen[key] = len(out)
|
|
||||||
out.append(row)
|
|
||||||
if dropped:
|
|
||||||
LOGGER.info(
|
|
||||||
"staging dedupe: %s collapsed %d duplicate rows within batch (kept last)",
|
|
||||||
dataset,
|
|
||||||
dropped,
|
|
||||||
extra={"hub": {"dataset": dataset, "deduped": dropped}},
|
|
||||||
)
|
|
||||||
return out
|
|
||||||
|
|
||||||
EOD_COPY = {
|
EOD_COPY = {
|
||||||
"stocks": (
|
"stocks": (
|
||||||
"INSERT OR REPLACE INTO eod_stocks "
|
"INSERT OR REPLACE INTO eod_stocks "
|
||||||
@@ -1491,20 +1439,8 @@ class Pipeline:
|
|||||||
else:
|
else:
|
||||||
keys = [(row.get("ts_code"), row.get("trade_date")) for row in rows]
|
keys = [(row.get("ts_code"), row.get("trade_date")) for row in rows]
|
||||||
dup = row_n - len(set(keys))
|
dup = row_n - len(set(keys))
|
||||||
# Extended soft datasets (popularity/dragon_tiger/…) already collapse
|
|
||||||
# within-batch dups in _dedupe_staging_rows before INSERT. Upstream
|
|
||||||
# ths/dc (and similar) routinely emit duplicate business keys; those
|
|
||||||
# collapsed dups must not hard-fail publish (HEL-562). Datasets without
|
|
||||||
# a staging business key — and all hard/core soft gates — still treat
|
|
||||||
# raw duplicate keys as errors.
|
|
||||||
staging_collapses_dups = (
|
|
||||||
dataset in EXTENDED_SOFT_DATASETS and dataset in STAGING_KEY_FIELDS
|
|
||||||
)
|
|
||||||
if dup:
|
if dup:
|
||||||
if staging_collapses_dups:
|
errors.append(f"duplicate keys: {dup}")
|
||||||
warnings.append(f"duplicate keys: {dup}")
|
|
||||||
else:
|
|
||||||
errors.append(f"duplicate keys: {dup}")
|
|
||||||
bad_date = sum(1 for row in rows if str(row.get("trade_date")) != trade_date)
|
bad_date = sum(1 for row in rows if str(row.get("trade_date")) != trade_date)
|
||||||
if bad_date:
|
if bad_date:
|
||||||
errors.append(f"date mismatch rows: {bad_date}")
|
errors.append(f"date mismatch rows: {bad_date}")
|
||||||
@@ -1523,8 +1459,7 @@ class Pipeline:
|
|||||||
field_report = self._field_gate(dataset, trade_date, rows, errors)
|
field_report = self._field_gate(dataset, trade_date, rows, errors)
|
||||||
if dataset in SOFT_DATASETS:
|
if dataset in SOFT_DATASETS:
|
||||||
allow_empty = dataset in {"popularity", "dragon_tiger", "moneyflow", "auction"}
|
allow_empty = dataset in {"popularity", "dragon_tiger", "moneyflow", "auction"}
|
||||||
hard_dup = 0 if staging_collapses_dups else dup
|
hard_fail = bool(dup or bad_date or (empty and not allow_empty))
|
||||||
hard_fail = bool(hard_dup or bad_date or (empty and not allow_empty))
|
|
||||||
else:
|
else:
|
||||||
hard_fail = bool(errors) and (dataset in HARD_DATASETS or dataset == STOCKS_DATASET)
|
hard_fail = bool(errors) and (dataset in HARD_DATASETS or dataset == STOCKS_DATASET)
|
||||||
report = {
|
report = {
|
||||||
@@ -1785,7 +1720,6 @@ class Pipeline:
|
|||||||
|
|
||||||
def _stage(self, dataset: str, batch_id: str, rows: list[dict[str, Any]]) -> None:
|
def _stage(self, dataset: str, batch_id: str, rows: list[dict[str, Any]]) -> None:
|
||||||
sql, mapper = STAGING_INSERT[dataset]
|
sql, mapper = STAGING_INSERT[dataset]
|
||||||
rows = _dedupe_staging_rows(dataset, rows)
|
|
||||||
with self.db.write() as connection:
|
with self.db.write() as connection:
|
||||||
connection.execute(
|
connection.execute(
|
||||||
f"DELETE FROM {DATASET_TABLES[dataset][1]} WHERE batch_id = ?",
|
f"DELETE FROM {DATASET_TABLES[dataset][1]} WHERE batch_id = ?",
|
||||||
|
|||||||
@@ -31,21 +31,21 @@ CATALOG: list[dict[str, Any]] = [
|
|||||||
"credential_key": "tushare_token",
|
"credential_key": "tushare_token",
|
||||||
"status_source": "src_health (legacy, kept) + provider_health (unified, HEL-543)",
|
"status_source": "src_health (legacy, kept) + provider_health (unified, HEL-543)",
|
||||||
"interfaces": [
|
"interfaces": [
|
||||||
{"interface": "trade_cal", "capability": "交易日历", "datasets": ["calendar"], "group": "日历 / 主档"},
|
{"interface": "trade_cal", "capability": "交易日历", "datasets": ["calendar"]},
|
||||||
{"interface": "stock_basic", "capability": "股票主档", "datasets": ["stocks"], "group": "日历 / 主档"},
|
{"interface": "stock_basic", "capability": "股票主档", "datasets": ["stocks"]},
|
||||||
{"interface": "daily", "capability": "个股日K", "datasets": ["daily"], "group": "盘后 A 批"},
|
{"interface": "daily", "capability": "个股日K", "datasets": ["daily"]},
|
||||||
{"interface": "adj_factor", "capability": "复权因子", "datasets": ["daily"], "group": "盘后 A 批"},
|
{"interface": "adj_factor", "capability": "复权因子", "datasets": ["daily"]},
|
||||||
{"interface": "daily_basic", "capability": "估值", "datasets": ["valuation"], "group": "盘后 A 批"},
|
{"interface": "daily_basic", "capability": "估值", "datasets": ["valuation"]},
|
||||||
{"interface": "index_daily", "capability": "指数日K", "datasets": ["index_daily"], "group": "指数 B 批"},
|
{"interface": "index_daily", "capability": "指数日K", "datasets": ["index_daily"]},
|
||||||
{"interface": "moneyflow", "capability": "资金流", "datasets": ["moneyflow"], "group": "盘后 A 批"},
|
{"interface": "moneyflow", "capability": "资金流", "datasets": ["moneyflow"]},
|
||||||
{"interface": "stk_auction", "capability": "集合竞价", "datasets": ["auction"], "group": "盘后 A 批"},
|
{"interface": "stk_auction", "capability": "集合竞价", "datasets": ["auction"]},
|
||||||
{"interface": "limit_list_d", "capability": "涨跌停池", "datasets": ["limit_events"], "group": "扩展软批"},
|
{"interface": "limit_list_d", "capability": "涨跌停池", "datasets": ["limit_events"]},
|
||||||
{"interface": "ths_hot", "capability": "同花顺人气榜", "datasets": ["popularity"], "group": "扩展软批"},
|
{"interface": "ths_hot", "capability": "同花顺人气榜", "datasets": ["popularity"]},
|
||||||
{"interface": "dc_hot", "capability": "东方财富人气榜", "datasets": ["popularity"], "group": "扩展软批"},
|
{"interface": "dc_hot", "capability": "东方财富人气榜", "datasets": ["popularity"]},
|
||||||
{"interface": "hm_detail", "capability": "龙虎榜游资明细", "datasets": ["dragon_tiger"], "group": "扩展软批"},
|
{"interface": "hm_detail", "capability": "龙虎榜游资明细", "datasets": ["dragon_tiger"]},
|
||||||
{"interface": "ths_daily", "capability": "同花顺概念行情", "datasets": ["sector_daily"], "group": "扩展软批"},
|
{"interface": "ths_daily", "capability": "同花顺概念行情", "datasets": ["sector_daily"]},
|
||||||
{"interface": "dc_index", "capability": "东方财富板块行情", "datasets": ["sector_daily"], "group": "扩展软批"},
|
{"interface": "dc_index", "capability": "东方财富板块行情", "datasets": ["sector_daily"]},
|
||||||
{"interface": "sw_daily", "capability": "申万行业行情", "datasets": ["sector_daily"], "group": "扩展软批"},
|
{"interface": "sw_daily", "capability": "申万行业行情", "datasets": ["sector_daily"]},
|
||||||
],
|
],
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
@@ -55,13 +55,13 @@ CATALOG: list[dict[str, Any]] = [
|
|||||||
"credential_key": None,
|
"credential_key": None,
|
||||||
"status_source": "provider_health (unified, HEL-543)",
|
"status_source": "provider_health (unified, HEL-543)",
|
||||||
"interfaces": [
|
"interfaces": [
|
||||||
{"interface": "indices", "capability": "指数实时报价", "datasets": ["index_quotes"], "group": "实时快照"},
|
{"interface": "indices", "capability": "指数实时报价", "datasets": ["index_quotes"]},
|
||||||
{"interface": "market_quotes", "capability": "全市场实时快照", "datasets": ["quotes_latest"], "group": "实时快照"},
|
{"interface": "market_quotes", "capability": "全市场实时快照", "datasets": ["quotes_latest"]},
|
||||||
{"interface": "named_quotes", "capability": "指定个股实时报价", "datasets": ["quotes_latest"], "group": "实时快照"},
|
{"interface": "named_quotes", "capability": "指定个股实时报价", "datasets": ["quotes_latest"]},
|
||||||
{"interface": "sector_quote", "capability": "申万板块实时报价(单个)", "datasets": ["sectors_quote"], "group": "实时快照"},
|
{"interface": "sector_quote", "capability": "申万板块实时报价(单个)", "datasets": ["sectors_quote"]},
|
||||||
{"interface": "sector_quotes_batch", "capability": "申万板块批量报价(预热)", "datasets": ["sectors_quote"], "group": "实时快照"},
|
{"interface": "sector_quotes_batch", "capability": "申万板块批量报价(预热)", "datasets": ["sectors_quote"]},
|
||||||
{"interface": "limit_pool", "capability": "涨停/炸板池(盘中)", "datasets": ["limit_pool"], "group": "实时快照"},
|
{"interface": "limit_pool", "capability": "涨停/炸板池(盘中)", "datasets": ["limit_pool"]},
|
||||||
{"interface": "intraday", "capability": "分时走势", "datasets": ["intraday_points"], "group": "实时快照"},
|
{"interface": "intraday", "capability": "分时走势", "datasets": ["intraday_points"]},
|
||||||
],
|
],
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
@@ -71,14 +71,13 @@ CATALOG: list[dict[str, Any]] = [
|
|||||||
"credential_key": None,
|
"credential_key": None,
|
||||||
"status_source": "provider_health (unified, HEL-543)",
|
"status_source": "provider_health (unified, HEL-543)",
|
||||||
"interfaces": [
|
"interfaces": [
|
||||||
{"interface": "indices", "capability": "指数实时报价(东财失败时备用)", "datasets": ["index_quotes"], "group": "实时备援"},
|
{"interface": "indices", "capability": "指数实时报价(东财失败时备用)", "datasets": ["index_quotes"]},
|
||||||
{
|
{
|
||||||
"interface": "market_quotes_fallback",
|
"interface": "market_quotes_fallback",
|
||||||
"capability": "全市场快照(备用；按本地股票主档逐只请求拼接)",
|
"capability": "全市场快照(备用；按本地股票主档逐只请求拼接)",
|
||||||
"datasets": ["quotes_latest"],
|
"datasets": ["quotes_latest"],
|
||||||
"group": "实时备援",
|
|
||||||
},
|
},
|
||||||
{"interface": "named_quotes", "capability": "指定个股实时报价(东财失败时备用)", "datasets": ["quotes_latest"], "group": "实时备援"},
|
{"interface": "named_quotes", "capability": "指定个股实时报价(东财失败时备用)", "datasets": ["quotes_latest"]},
|
||||||
],
|
],
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
@@ -88,11 +87,11 @@ CATALOG: list[dict[str, Any]] = [
|
|||||||
"credential_key": "ifind_refresh_token",
|
"credential_key": "ifind_refresh_token",
|
||||||
"status_source": "provider_health (unified, HEL-543) + adapter.status()",
|
"status_source": "provider_health (unified, HEL-543) + adapter.status()",
|
||||||
"interfaces": [
|
"interfaces": [
|
||||||
{"interface": "wencai", "capability": "问财自然语言选股", "datasets": ["ifind_wencai"], "group": "预留接口"},
|
{"interface": "wencai", "capability": "问财自然语言选股", "datasets": ["ifind_wencai"]},
|
||||||
{"interface": "snapshots", "capability": "快照", "datasets": ["ifind_snapshots"], "group": "预留接口"},
|
{"interface": "snapshots", "capability": "快照", "datasets": ["ifind_snapshots"]},
|
||||||
{"interface": "history", "capability": "历史行情", "datasets": ["ifind_history"], "group": "预留接口"},
|
{"interface": "history", "capability": "历史行情", "datasets": ["ifind_history"]},
|
||||||
{"interface": "realtime", "capability": "实时行情", "datasets": ["ifind_realtime"], "group": "预留接口"},
|
{"interface": "realtime", "capability": "实时行情", "datasets": ["ifind_realtime"]},
|
||||||
{"interface": "intraday", "capability": "分时(高频)", "datasets": ["ifind_intraday"], "group": "预留接口"},
|
{"interface": "intraday", "capability": "分时(高频)", "datasets": ["ifind_intraday"]},
|
||||||
],
|
],
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
@@ -162,35 +161,5 @@ def snapshot(db: Any, auth: Any = None) -> list[dict[str, Any]]:
|
|||||||
except Exception:
|
except Exception:
|
||||||
health_rows = []
|
health_rows = []
|
||||||
item["live_interfaces"] = health_rows
|
item["live_interfaces"] = health_rows
|
||||||
# HEL-529 fix 3: 已登记 ≠ 已观测 ≠ 健康。观测记录有两种真实写法：
|
|
||||||
# eastmoney/tencent 的 observe() 直接写接口名；tushare 官方管线
|
|
||||||
# _log_call() 写的是数据集名（如 daily_basic 接口对应的数据集
|
|
||||||
# valuation）。这里按「接口名 或 该接口声明的 datasets 之一」双向
|
|
||||||
# 匹配，并为每个接口标注观测依据，禁止把"暂无观测"显示成"未配置"。
|
|
||||||
by_interface = {str(row["interface"]): row for row in health_rows}
|
|
||||||
for iface in item["interfaces"]:
|
|
||||||
row = by_interface.get(str(iface["interface"]))
|
|
||||||
basis = "interface" if row is not None else ""
|
|
||||||
if row is None:
|
|
||||||
for dataset in iface.get("datasets", []):
|
|
||||||
candidate = by_interface.get(str(dataset))
|
|
||||||
if candidate is not None:
|
|
||||||
row = candidate
|
|
||||||
basis = "dataset"
|
|
||||||
break
|
|
||||||
if row is None:
|
|
||||||
iface["observed"] = False
|
|
||||||
iface["observed_basis"] = ""
|
|
||||||
iface["observed_state"] = ""
|
|
||||||
iface["observed_at"] = ""
|
|
||||||
iface["observed_latency_ms"] = None
|
|
||||||
iface["observed_note"] = ""
|
|
||||||
else:
|
|
||||||
iface["observed"] = True
|
|
||||||
iface["observed_basis"] = basis
|
|
||||||
iface["observed_state"] = str(row["state"] or "")
|
|
||||||
iface["observed_at"] = str(row["last_ok_at"] or row["updated_at"] or "")
|
|
||||||
iface["observed_latency_ms"] = row["last_latency_ms"]
|
|
||||||
iface["observed_note"] = str(row["last_error"] or row["last_fallback_reason"] or "")
|
|
||||||
result.append(item)
|
result.append(item)
|
||||||
return result
|
return result
|
||||||
|
|||||||
@@ -1,345 +0,0 @@
|
|||||||
"""HEL-529 rework regressions: staging dedupe, anomaly convergence,
|
|
||||||
source-catalog observation join (dataset-name ↔ interface-name), lineage
|
|
||||||
update_freq. All read-only or within-batch fixes; none touch routing, the
|
|
||||||
8765 main site, or the 问天 frozen zone.
|
|
||||||
"""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import tempfile
|
|
||||||
import unittest
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from datahub.adapters.tushare import TushareAdapter
|
|
||||||
from datahub.crypto import SecretVault
|
|
||||||
from datahub.hub import Hub
|
|
||||||
from datahub.pipeline import _dedupe_staging_rows
|
|
||||||
from datahub.settings import Settings
|
|
||||||
from tests.fixtures import fake_transport
|
|
||||||
|
|
||||||
|
|
||||||
class StagingDedupeTests(unittest.TestCase):
|
|
||||||
def test_popularity_within_batch_duplicates_collapse_keep_last(self) -> None:
|
|
||||||
rows = [
|
|
||||||
{"ts_code": "600000.SH", "trade_date": "20260914", "source": "ths", "rank": 1},
|
|
||||||
{"ts_code": "000868.SZ", "trade_date": "20260914", "source": "dc", "rank": 2},
|
|
||||||
{"ts_code": "600000.SH", "trade_date": "20260914", "source": "dc", "rank": 3, "hot": 9.9},
|
|
||||||
{"ts_code": "600000.SH", "trade_date": "20260914", "source": "dc", "rank": 4, "hot": 8.8},
|
|
||||||
]
|
|
||||||
out = _dedupe_staging_rows("popularity", rows)
|
|
||||||
self.assertEqual(len(out), 3) # (600000,ths) (000868,dc) (600000,dc)
|
|
||||||
dup = [r for r in out if r["ts_code"] == "600000.SH" and r["source"] == "dc"][0]
|
|
||||||
self.assertEqual(dup["rank"], 4) # keeps LAST occurrence
|
|
||||||
self.assertEqual(out[0]["ts_code"], "600000.SH") # preserves first-seen order
|
|
||||||
|
|
||||||
def test_dragon_tiger_seat_duplicates_collapse(self) -> None:
|
|
||||||
rows = [
|
|
||||||
{"ts_code": "300010.SZ", "trade_date": "20260914", "hm_name": "T王", "buy_amount": 100},
|
|
||||||
{"ts_code": "300010.SZ", "trade_date": "20260914", "hm_name": "T王", "buy_amount": 200},
|
|
||||||
{"ts_code": "300010.SZ", "trade_date": "20260914", "hm_name": "T王", "buy_amount": 300},
|
|
||||||
]
|
|
||||||
out = _dedupe_staging_rows("dragon_tiger", rows)
|
|
||||||
self.assertEqual(len(out), 1)
|
|
||||||
self.assertEqual(out[0]["buy_amount"], 300)
|
|
||||||
|
|
||||||
def test_different_sources_are_not_duplicates(self) -> None:
|
|
||||||
rows = [
|
|
||||||
{"ts_code": "600000.SH", "trade_date": "20260914", "source": "ths"},
|
|
||||||
{"ts_code": "600000.SH", "trade_date": "20260914", "source": "dc"},
|
|
||||||
]
|
|
||||||
self.assertEqual(len(_dedupe_staging_rows("popularity", rows)), 2)
|
|
||||||
|
|
||||||
def test_unknown_dataset_passthrough(self) -> None:
|
|
||||||
rows = [{"a": 1}, {"a": 1}]
|
|
||||||
self.assertEqual(_dedupe_staging_rows("calendar", rows), rows)
|
|
||||||
|
|
||||||
|
|
||||||
class _Base(unittest.TestCase):
|
|
||||||
def setUp(self) -> None:
|
|
||||||
self.tmp = tempfile.TemporaryDirectory()
|
|
||||||
settings = Settings(
|
|
||||||
encryption_key=SecretVault.generate_key(),
|
|
||||||
api_token="z" * 32,
|
|
||||||
admin_password="StartPass1",
|
|
||||||
tushare_token="real-tushare-token-abcdef",
|
|
||||||
db_path=Path(self.tmp.name) / "hub.db",
|
|
||||||
scheduler_enabled=False,
|
|
||||||
)
|
|
||||||
self.hub = Hub(settings, adapter=TushareAdapter("real-tushare-token-abcdef", transport=fake_transport))
|
|
||||||
|
|
||||||
def tearDown(self) -> None:
|
|
||||||
self.tmp.cleanup()
|
|
||||||
|
|
||||||
|
|
||||||
class StagingDedupePublishTests(_Base):
|
|
||||||
def test_duplicate_popularity_and_dragon_tiger_now_publish(self) -> None:
|
|
||||||
"""Replays the 2026-09-14 production failure: within-response duplicate
|
|
||||||
keys used to abort the whole batch at the staging INSERT; with dedupe
|
|
||||||
the same upstream payload publishes."""
|
|
||||||
db = self.hub.db
|
|
||||||
trade_date = "20240902"
|
|
||||||
# dc_hot returns 600000.SH twice within one response; hm_detail returns
|
|
||||||
# the same (ts_code, hm_name) seat three times (mirrors live evidence).
|
|
||||||
popularity_rows = [
|
|
||||||
{"ts_code": "600000.SH", "trade_date": trade_date, "source": "ths",
|
|
||||||
"ts_name": "浦发银行", "rank": 1, "pct_change": 1.2, "current_price": 10.2,
|
|
||||||
"hot": 90.0, "concept": "银行", "data_type": "热股"},
|
|
||||||
{"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc",
|
|
||||||
"ts_name": "浦发银行", "rank": 2, "pct_change": 1.2, "current_price": 10.2,
|
|
||||||
"hot": 80.0, "concept": "银行", "data_type": "A股市场"},
|
|
||||||
{"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc",
|
|
||||||
"ts_name": "浦发银行", "rank": 3, "pct_change": 1.3, "current_price": 10.3,
|
|
||||||
"hot": 81.0, "concept": "银行", "data_type": "A股市场"},
|
|
||||||
]
|
|
||||||
dragon_rows = [
|
|
||||||
{"trade_date": trade_date, "ts_code": "600000.SH", "ts_name": "浦发银行",
|
|
||||||
"buy_amount": 100, "sell_amount": 200, "net_amount": -100,
|
|
||||||
"hm_name": "测试游资", "hm_orgs": "某某营业部", "tag": "超买"},
|
|
||||||
{"trade_date": trade_date, "ts_code": "600000.SH", "ts_name": "浦发银行",
|
|
||||||
"buy_amount": 300, "sell_amount": 0, "net_amount": 300,
|
|
||||||
"hm_name": "测试游资", "hm_orgs": "某某营业部", "tag": "超买"},
|
|
||||||
]
|
|
||||||
self.hub.pipeline._stage("popularity", "b-dup-pop", popularity_rows)
|
|
||||||
self.hub.pipeline._stage("dragon_tiger", "b-dup-dt", dragon_rows)
|
|
||||||
pop = db.fetchall("SELECT * FROM staging_popularity WHERE batch_id = 'b-dup-pop'")
|
|
||||||
dt = db.fetchall("SELECT * FROM staging_dragon_tiger WHERE batch_id = 'b-dup-dt'")
|
|
||||||
self.assertEqual(len(pop), 2)
|
|
||||||
self.assertEqual(len(dt), 1)
|
|
||||||
kept = dt[0]
|
|
||||||
self.assertEqual(kept["buy_amount"], 300)
|
|
||||||
|
|
||||||
def test_hel562_popularity_collapsed_dups_publish_not_hard_fail(self) -> None:
|
|
||||||
"""HEL-562: staging dedupe alone is not enough — quality gate used to
|
|
||||||
hard-fail on the same within-batch dups after they were already
|
|
||||||
collapsed (live 20260915: duplicate keys: 3 → integrity_gate)."""
|
|
||||||
trade_date = "20240902"
|
|
||||||
# 3 within-batch dups on (ts_code, trade_date, source=dc) — mirrors
|
|
||||||
# ths+dc merge where dc_hot repeats the same keys.
|
|
||||||
rows = [
|
|
||||||
{"ts_code": "600000.SH", "trade_date": trade_date, "source": "ths",
|
|
||||||
"ts_name": "浦发银行", "rank": 1, "pct_change": 1.2, "current_price": 10.2,
|
|
||||||
"hot": 90.0, "concept": "银行", "data_type": "热股"},
|
|
||||||
{"ts_code": "000001.SZ", "trade_date": trade_date, "source": "dc",
|
|
||||||
"ts_name": "平安银行", "rank": 1, "pct_change": 2.0, "current_price": 11.0,
|
|
||||||
"hot": 88.0, "concept": "银行", "data_type": "A股市场"},
|
|
||||||
{"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc",
|
|
||||||
"ts_name": "浦发银行", "rank": 2, "pct_change": 1.2, "current_price": 10.2,
|
|
||||||
"hot": 80.0, "concept": "银行", "data_type": "A股市场"},
|
|
||||||
{"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc",
|
|
||||||
"ts_name": "浦发银行", "rank": 3, "pct_change": 1.3, "current_price": 10.3,
|
|
||||||
"hot": 81.0, "concept": "银行", "data_type": "A股市场"},
|
|
||||||
{"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc",
|
|
||||||
"ts_name": "浦发银行", "rank": 4, "pct_change": 1.4, "current_price": 10.4,
|
|
||||||
"hot": 82.0, "concept": "银行", "data_type": "A股市场"},
|
|
||||||
]
|
|
||||||
report = self.hub.pipeline.validate("popularity", "b-gate", trade_date, rows)
|
|
||||||
self.assertFalse(report["hard_fail"])
|
|
||||||
self.assertEqual(report["errors"], [])
|
|
||||||
self.assertEqual(report["warnings"], ["duplicate keys: 2"])
|
|
||||||
|
|
||||||
result = self.hub.pipeline.run_dataset("popularity", trade_date, prepared_rows=rows)
|
|
||||||
# warnings present → soft_fail → publication state is degraded (still served)
|
|
||||||
self.assertEqual(result["state"], "degraded")
|
|
||||||
self.assertFalse(result["quality"]["hard_fail"])
|
|
||||||
self.assertTrue(result["quality"]["soft_fail"])
|
|
||||||
self.assertIn("duplicate keys: 2", result["quality"]["warnings"])
|
|
||||||
eod = self.hub.db.fetchall(
|
|
||||||
"SELECT * FROM eod_popularity WHERE trade_date = ? AND batch_id = ?",
|
|
||||||
(trade_date, result["batch_id"]),
|
|
||||||
)
|
|
||||||
# 5 raw → 3 unique keys after staging collapse (ths + two dc codes)
|
|
||||||
self.assertEqual(len(eod), 3)
|
|
||||||
pub = self.hub.db.fetchone(
|
|
||||||
"SELECT * FROM publications WHERE dataset='popularity' AND trade_date=?",
|
|
||||||
(trade_date,),
|
|
||||||
)
|
|
||||||
self.assertEqual(pub["active_batch"], result["batch_id"])
|
|
||||||
self.assertEqual(pub["state"], "degraded")
|
|
||||||
# Serving path accepts degraded the same as published (no DATASET_NOT_PUBLISHED)
|
|
||||||
from datahub.serving import V1API
|
|
||||||
|
|
||||||
api = V1API(self.hub.db, self.hub.pipeline, self.hub.settings)
|
|
||||||
payload = api.handle("/v1/popularity", {"date": [trade_date]})
|
|
||||||
self.assertEqual(len(payload["data"]), 3)
|
|
||||||
self.assertEqual(payload["meta"]["state"], "degraded")
|
|
||||||
self.assertEqual(payload["meta"]["batch_id"], result["batch_id"])
|
|
||||||
|
|
||||||
def test_hel562_core_soft_still_hard_fails_on_duplicate_keys(self) -> None:
|
|
||||||
"""moneyflow/auction stay on the old soft gate: raw dups → hard_fail."""
|
|
||||||
trade_date = "20240902"
|
|
||||||
rows = [
|
|
||||||
{"ts_code": "600000.SH", "trade_date": trade_date,
|
|
||||||
"buy_sm_amount": 1, "sell_sm_amount": 1, "buy_md_amount": 1, "sell_md_amount": 1,
|
|
||||||
"buy_lg_amount": 1, "sell_lg_amount": 1, "buy_elg_amount": 1, "sell_elg_amount": 1,
|
|
||||||
"net_mf_amount": 0},
|
|
||||||
{"ts_code": "600000.SH", "trade_date": trade_date,
|
|
||||||
"buy_sm_amount": 2, "sell_sm_amount": 2, "buy_md_amount": 2, "sell_md_amount": 2,
|
|
||||||
"buy_lg_amount": 2, "sell_lg_amount": 2, "buy_elg_amount": 2, "sell_elg_amount": 2,
|
|
||||||
"net_mf_amount": 0},
|
|
||||||
]
|
|
||||||
report = self.hub.pipeline.validate("moneyflow", "b-mf", trade_date, rows)
|
|
||||||
self.assertTrue(report["hard_fail"])
|
|
||||||
self.assertIn("duplicate keys: 1", report["errors"])
|
|
||||||
self.assertEqual(report["warnings"], [])
|
|
||||||
|
|
||||||
def test_hel562_popularity_date_mismatch_still_hard_fails(self) -> None:
|
|
||||||
"""Collapsed-dup carve-out must not weaken other soft integrity checks."""
|
|
||||||
rows = [
|
|
||||||
{"ts_code": "600000.SH", "trade_date": "20240901", "source": "ths",
|
|
||||||
"ts_name": "浦发银行", "rank": 1, "pct_change": 1.2, "current_price": 10.2,
|
|
||||||
"hot": 90.0, "concept": "银行", "data_type": "热股"},
|
|
||||||
]
|
|
||||||
report = self.hub.pipeline.validate("popularity", "b-bad-date", "20240902", rows)
|
|
||||||
self.assertTrue(report["hard_fail"])
|
|
||||||
self.assertIn("date mismatch rows: 1", report["errors"])
|
|
||||||
|
|
||||||
|
|
||||||
class OverviewAnomalyConvergenceTests(_Base):
|
|
||||||
def _seed_batches(self, today: str) -> None:
|
|
||||||
db = self.hub.db
|
|
||||||
rows = [
|
|
||||||
# resolved history: earlier failures/stalls, later success
|
|
||||||
("x-daily-001", "daily", "staged", "empty official batch", "2026-09-14T15:05:00+08:00"),
|
|
||||||
("x-daily-002", "daily", "failed", "release group not switched", "2026-09-14T16:10:00+08:00"),
|
|
||||||
("x-daily-006", "daily", "published", "", "2026-09-14T20:00:00+08:00"),
|
|
||||||
# current unresolved faults
|
|
||||||
("x-pop-001", "popularity", "failed", "UNIQUE constraint failed: staging_popularity", "2026-09-14T22:40:00+08:00"),
|
|
||||||
("x-dt-001", "dragon_tiger", "failed", "UNIQUE constraint failed: staging_dragon_tiger", "2026-09-14T16:45:00+08:00"),
|
|
||||||
("x-dt-002", "dragon_tiger", "failed", "UNIQUE constraint failed: staging_dragon_tiger", "2026-09-14T21:48:00+08:00"),
|
|
||||||
# staged-empty later published
|
|
||||||
("x-idx-001", "index_daily", "staged", "empty official batch", "2026-09-14T15:10:00+08:00"),
|
|
||||||
("x-idx-003", "index_daily", "published", "", "2026-09-14T16:10:00+08:00"),
|
|
||||||
]
|
|
||||||
for batch_id, dataset, state, error, started in rows:
|
|
||||||
db.execute(
|
|
||||||
"INSERT INTO batches(batch_id, dataset, trade_date, state, attempt, rows_in, rows_out,"
|
|
||||||
" quality_json, started_at, finished_at, error) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
|
|
||||||
(batch_id, dataset, today, state, 1, None, None, None, started, started if state != "staged" else None, error or None),
|
|
||||||
)
|
|
||||||
db.execute(
|
|
||||||
"INSERT INTO publications(dataset, trade_date, active_batch, prev_batch, state, published_at)"
|
|
||||||
" VALUES ('daily', ?, 'x-daily-006', 'x-daily-005', 'published', '2026-09-14T20:00:22+08:00')",
|
|
||||||
(today,),
|
|
||||||
)
|
|
||||||
db.execute(
|
|
||||||
"INSERT INTO publications(dataset, trade_date, active_batch, prev_batch, state, published_at)"
|
|
||||||
" VALUES ('index_daily', ?, 'x-idx-003', 'x-idx-002', 'published', '2026-09-14T16:10:12+08:00')",
|
|
||||||
(today,),
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_anomalies_only_latest_unresolved(self) -> None:
|
|
||||||
overview = self.hub.admin.overview()
|
|
||||||
today = overview["trade_date"]
|
|
||||||
self._seed_batches(today)
|
|
||||||
anomalies = self.hub.admin.overview()["anomalies"]
|
|
||||||
got = sorted((a["dataset"], a["batch_id"]) for a in anomalies)
|
|
||||||
self.assertEqual(
|
|
||||||
got,
|
|
||||||
[
|
|
||||||
("dragon_tiger", "x-dt-002"), # latest failed, never published
|
|
||||||
("popularity", "x-pop-001"), # latest failed, never published
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class SourceCatalogJoinTests(_Base):
|
|
||||||
def test_observation_join_matches_dataset_named_health_rows(self) -> None:
|
|
||||||
db = self.hub.db
|
|
||||||
now = "2026-09-15T08:00:00+08:00"
|
|
||||||
# tushare observability writes dataset names (real legacy behavior)
|
|
||||||
for iface, state in [("valuation", "ok"), ("popularity", "ok"), ("stocks", "ok")]:
|
|
||||||
db.execute(
|
|
||||||
"INSERT INTO provider_health(provider, interface, state, last_ok_at, last_error,"
|
|
||||||
" last_fallback_reason, consec_failures, last_latency_ms, last_data_age_seconds, updated_at)"
|
|
||||||
" VALUES ('tushare', ?, ?, ?, '', '', 0, 300, NULL, ?)",
|
|
||||||
(iface, state, now, now),
|
|
||||||
)
|
|
||||||
# realtime providers write interface names
|
|
||||||
db.execute(
|
|
||||||
"INSERT INTO provider_health(provider, interface, state, last_ok_at, last_error,"
|
|
||||||
" last_fallback_reason, consec_failures, last_latency_ms, last_data_age_seconds, updated_at)"
|
|
||||||
" VALUES ('eastmoney', 'indices', 'ok', ?, '', '', 0, 153, NULL, ?)",
|
|
||||||
(now, now),
|
|
||||||
)
|
|
||||||
items = self.hub.admin.source_catalog()["items"]
|
|
||||||
tushare = [i for i in items if i["provider"] == "tushare"][0]
|
|
||||||
by_iface = {i["interface"]: i for i in tushare["interfaces"]}
|
|
||||||
# daily_basic serves valuation → observed via dataset name
|
|
||||||
self.assertTrue(by_iface["daily_basic"]["observed"])
|
|
||||||
self.assertEqual(by_iface["daily_basic"]["observed_basis"], "dataset")
|
|
||||||
self.assertEqual(by_iface["daily_basic"]["observed_state"], "ok")
|
|
||||||
# ths_hot + dc_hot serve popularity → observed via dataset name
|
|
||||||
self.assertTrue(by_iface["ths_hot"]["observed"])
|
|
||||||
self.assertTrue(by_iface["dc_hot"]["observed"])
|
|
||||||
# stock_basic serves stocks
|
|
||||||
self.assertTrue(by_iface["stock_basic"]["observed"])
|
|
||||||
# never-observed interface stays honestly unobserved (not "unconfigured")
|
|
||||||
self.assertFalse(by_iface["trade_cal"]["observed"])
|
|
||||||
self.assertEqual(by_iface["trade_cal"]["observed_state"], "")
|
|
||||||
# interfaces carry real batch groups
|
|
||||||
groups = {i["interface"]: i["group"] for i in tushare["interfaces"]}
|
|
||||||
self.assertEqual(groups["daily"], "盘后 A 批")
|
|
||||||
self.assertEqual(groups["ths_hot"], "扩展软批")
|
|
||||||
self.assertEqual(groups["index_daily"], "指数 B 批")
|
|
||||||
eastmoney = [i for i in items if i["provider"] == "eastmoney"][0]
|
|
||||||
em = {i["interface"]: i for i in eastmoney["interfaces"]}
|
|
||||||
self.assertTrue(em["indices"]["observed"])
|
|
||||||
self.assertEqual(em["indices"]["observed_basis"], "interface")
|
|
||||||
self.assertFalse(em["market_quotes"]["observed"])
|
|
||||||
|
|
||||||
def test_lineage_update_freq_present(self) -> None:
|
|
||||||
items = self.hub.admin.lineage("20240902")["items"]
|
|
||||||
self.assertTrue(items)
|
|
||||||
for item in items:
|
|
||||||
self.assertTrue(item.get("update_freq"), f"missing update_freq for {item['dataset']}")
|
|
||||||
|
|
||||||
|
|
||||||
class LineageMentorIfindTests(_Base):
|
|
||||||
"""问师 → iFinD 血缘修正（第二轮返工）：不按名称猜关系，按真实调用代码。"""
|
|
||||||
|
|
||||||
REPO = Path(__file__).resolve().parents[2]
|
|
||||||
|
|
||||||
def test_no_mentor_dependency_on_ifind_wencai(self) -> None:
|
|
||||||
items = self.hub.admin.lineage("20240902")["items"]
|
|
||||||
wencai = [i for i in items if i["dataset"] == "ifind_wencai"]
|
|
||||||
self.assertEqual(len(wencai), 1)
|
|
||||||
consumers = wencai[0]["known_consumers"]
|
|
||||||
for consumer in consumers:
|
|
||||||
self.assertNotIn("问师", consumer, f"wencai consumer must not be 问师: {consumer}")
|
|
||||||
all_consumers = " ".join(c for i in items for c in i["known_consumers"])
|
|
||||||
self.assertNotIn("问师(自然语言选股", all_consumers)
|
|
||||||
|
|
||||||
def test_wencai_real_consumer_is_pools_enrichment_with_code_evidence(self) -> None:
|
|
||||||
items = self.hub.admin.lineage("20240902")["items"]
|
|
||||||
wencai = [i for i in items if i["dataset"] == "ifind_wencai"][0]
|
|
||||||
self.assertTrue(any("股票池" in c for c in wencai["known_consumers"]))
|
|
||||||
# Real call evidence in the main-site source tree:
|
|
||||||
pools_src = (self.REPO / "backend" / "features" / "pools" / "service.py").read_text(encoding="utf-8")
|
|
||||||
self.assertIn("ifind.wencai(", pools_src)
|
|
||||||
self.assertIn("ifind_event_enrichment_v1", pools_src)
|
|
||||||
# And 问师 itself never calls wencai:
|
|
||||||
mentor_src = (self.REPO / "backend" / "features" / "mentor" / "service.py").read_text(encoding="utf-8")
|
|
||||||
self.assertNotIn(".wencai(", mentor_src)
|
|
||||||
|
|
||||||
def test_mentor_optional_ifind_history_subcapabilities(self) -> None:
|
|
||||||
items = self.hub.admin.lineage("20240902")["items"]
|
|
||||||
history = [i for i in items if i["dataset"] == "ifind_history"]
|
|
||||||
self.assertEqual(len(history), 1)
|
|
||||||
consumers = history[0]["known_consumers"]
|
|
||||||
self.assertTrue(any("趋势思维模型" in c for c in consumers))
|
|
||||||
self.assertTrue(any("宏观思维模型" in c for c in consumers))
|
|
||||||
# every consumer must be a 问师 sub-capability, not the whole board
|
|
||||||
for consumer in consumers:
|
|
||||||
self.assertIn("·", consumer, f"not a sub-capability mapping: {consumer}")
|
|
||||||
# Real call evidence: mentor builds market matrices via ifind.history
|
|
||||||
mentor_src = (self.REPO / "backend" / "features" / "mentor" / "service.py").read_text(encoding="utf-8")
|
|
||||||
self.assertIn("ifind.history(", mentor_src)
|
|
||||||
self.assertIn("_mentor_market_matrix", mentor_src)
|
|
||||||
self.assertIn("MENTOR_INDEX_UNIVERSE", mentor_src)
|
|
||||||
self.assertIn("MENTOR_ETF_UNIVERSE", mentor_src)
|
|
||||||
# Optional dependency: fails open when ifind is not configured
|
|
||||||
self.assertIn("if not ifind or not ifind.configured", mentor_src)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
Reference in New Issue
Block a user