windows-desktop: C端基线（Windows 本地实时分析器 + 采集器 bug 修复）

test_fetch.py

@@ -0,0 +1,74 @@
+"""快速采集测试脚本"""
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).parent))
+
+from src.fetcher import stock_api, lhb_fetcher, news_flash_fetcher, macro_event_fetcher
+from src.storage.db import upsert_df, upsert_log, get_conn
+from datetime import datetime, timedelta
+
+def test_daily():
+    print("=== [1/4] 日K线（主要指数 90天） ===")
+    df = stock_api.batch_fetch_indices(days=90)
+    if df is not None and not df.empty:
+        upsert_df(df, "stock_daily")
+        print(f"  索引列表: {sorted(df['ts_code'].unique())}")
+        print(f"  总行数: {len(df)}")
+        # 验证最近一条
+        latest = df.sort_values("trade_date").tail(3)
+        print(f"  最新3条:\n{latest[['ts_code','trade_date','close']].to_string(index=False)}")
+    else:
+        print("  [FAIL] 返回空")
+
+def test_lhb():
+    print("\n=== [2/4] 龙虎榜（近10日，含上周五） ===")
+    # 上周五（最近确定有数据的交易日）
+    start = (datetime.today() - timedelta(days=10)).strftime("%Y-%m-%d")
+    end   = datetime.today().strftime("%Y-%m-%d")
+    df = lhb_fetcher.fetch_lhb_range(start, end)
+    if df is not None and not df.empty:
+        upsert_df(df, "lhb_daily")
+        upsert_log("lhb_daily", "", last_date=end)
+        print(f"  总行数: {len(df)}")
+        print(f"  涵盖交易日: {sorted(df['trade_date'].unique())}")
+        print(f"  样例:\n{df[['trade_date','ts_code','name','reason']].head(5).to_string(index=False)}")
+    else:
+        print("  [WARN] 该时间段无龙虎榜数据（可能是非交易日）")
+
+def test_flash():
+    print("\n=== [3/4] 7x24 快讯流（AkShare 财联社） ===")
+    df = news_flash_fetcher.fetch_akshare_news()
+    if df is not None and not df.empty:
+        upsert_df(df, "news_flash", csv备份=False)
+        print(f"  总行数: {len(df)}")
+        print(f"  样例:\n{df.head(3).to_string(index=False)}")
+    else:
+        print("  [WARN] 无快讯数据")
+
+def test_events():
+    print("\n=== [4/4] 大事提醒 ===")
+    df = macro_event_fetcher.fetch_macro_calendar()
+    if df is not None and not df.empty:
+        upsert_df(df, "macro_event")
+        print(f"  总行数: {len(df)}")
+        print(f"  样例:\n{df.head(5).to_string(index=False)}")
+    else:
+        print("  [WARN] 无数据（可能接口报表名需调整）")
+
+def db_summary():
+    print("\n=== 数据库现状 ===")
+    with get_conn() as conn:
+        tables = ["stock_daily","lhb_daily","news_flash","macro_event","news_cn"]
+        for t in tables:
+            try:
+                cnt = conn.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
+                print(f"  {t}: {cnt} rows")
+            except Exception:
+                print(f"  {t}: 0 rows (table may not exist)")
+
+if __name__ == "__main__":
+    test_daily()
+    test_lhb()
+    test_flash()
+    test_events()
+    db_summary()