75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
"""快速采集测试脚本"""
|
|
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()
|