songzhuoyuan
110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
数据源改造实盘验证脚本
|
|
请在【有网络】的环境运行（例如在 PyCharm 的 aiagents-stock 解释器中执行），
|
|
它会逐项调用各模块的数据获取接口，并打印 PASS/FAIL 与实际使用的数据源日志。
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
|
|
PROJECT = os.environ.get("AIAGENTS_STOCK_HOME", "/Users/songzhuoyuan/Desktop/code/python/aiagents-stock")
|
|
if not os.path.isdir(PROJECT):
|
|
PROJECT = input("请输入项目绝对路径: ").strip()
|
|
sys.path.insert(0, PROJECT)
|
|
os.chdir(PROJECT)
|
|
|
|
print("=" * 60)
|
|
print("数据源改造验证 - 股票代码统一使用 600637")
|
|
print("=" * 60)
|
|
|
|
from data_source_manager import data_source_manager
|
|
print("Tushare Token 已配置:", bool(data_source_manager.tushare_token))
|
|
print("Tushare 可用:", data_source_manager.tushare_available)
|
|
print()
|
|
|
|
passed = 0
|
|
failed = 0
|
|
|
|
def test(name, fn):
|
|
global passed, failed
|
|
try:
|
|
result = fn()
|
|
if result:
|
|
passed += 1
|
|
print(f"PASS | {name}")
|
|
else:
|
|
failed += 1
|
|
print(f"FAIL | {name}（返回空/失败）")
|
|
except Exception as e:
|
|
failed += 1
|
|
print(f"FAIL | {name}: {type(e).__name__}: {str(e)[:120]}")
|
|
|
|
# 1. 数据源管理器（核心）
|
|
test("历史数据(1年,前复权)", lambda: data_source_manager.get_stock_hist_data("600637", start_date="20250801", end_date="20260810", adjust="qfq") is not None)
|
|
test("基本信息", lambda: data_source_manager.get_stock_basic_info("600637").get("name") != "未知")
|
|
test("实时行情", lambda: bool(data_source_manager.get_realtime_quotes("600637")))
|
|
test("财务数据(利润表)", lambda: data_source_manager.get_financial_data("600637", "income") is not None)
|
|
|
|
# 2. 资金流向
|
|
def fund_flow_test():
|
|
from fund_flow_akshare import FundFlowAkshareDataFetcher
|
|
return FundFlowAkshareDataFetcher().get_fund_flow_data("600637").get("data_success")
|
|
test("资金流向", fund_flow_test)
|
|
|
|
# 3. 季报
|
|
def quarterly_test():
|
|
from quarterly_report_data import QuarterlyReportDataFetcher
|
|
return QuarterlyReportDataFetcher().get_quarterly_reports("600637").get("data_success")
|
|
test("季报(三表+指标)", quarterly_test)
|
|
|
|
# 4. 新闻
|
|
def news_test():
|
|
from qstock_news_data import QStockNewsDataFetcher
|
|
return QStockNewsDataFetcher().get_stock_news("600637").get("data_success")
|
|
test("个股新闻", news_test)
|
|
|
|
# 5. 市场情绪
|
|
def sentiment_tests():
|
|
from market_sentiment_data import MarketSentimentDataFetcher
|
|
f = MarketSentimentDataFetcher()
|
|
return (f._get_turnover_rate("600637") is not None and
|
|
f._get_market_index_sentiment() is not None)
|
|
test("情绪-换手率/大盘指数", sentiment_tests)
|
|
|
|
def limit_test():
|
|
from market_sentiment_data import MarketSentimentDataFetcher
|
|
return MarketSentimentDataFetcher()._get_limit_up_down_stats() is not None
|
|
test("情绪-涨跌停统计", limit_test)
|
|
|
|
def margin_test():
|
|
from market_sentiment_data import MarketSentimentDataFetcher
|
|
return MarketSentimentDataFetcher()._get_margin_trading_data("600637") is not None
|
|
test("情绪-融资融券", margin_test)
|
|
|
|
# 6. 综合股票数据（主分析链路）
|
|
def stock_data_test():
|
|
from stock_data import StockDataFetcher
|
|
f = StockDataFetcher()
|
|
info = f.get_stock_info("600637")
|
|
data = f.get_stock_data("600637", "1y")
|
|
fin = f.get_financial_data("600637")
|
|
return (isinstance(data, dict) and "error" not in data) and len(fin) > 0
|
|
test("主分析链路(信息/行情/财务)", stock_data_test)
|
|
|
|
# 7. 港股（可选，若积分不足会自动回退akshare）
|
|
def hk_test():
|
|
from stock_data import StockDataFetcher
|
|
f = StockDataFetcher()
|
|
data = f.get_stock_data("00700", "1mo")
|
|
return isinstance(data, dict) and "error" not in data
|
|
test("港股日线(00700)", hk_test)
|
|
|
|
print()
|
|
print("=" * 60)
|
|
print(f"验证完成：通过 {passed} 项，失败 {failed} 项")
|
|
if failed:
|
|
print("注意：失败项通常表示对应 tushare 接口积分不足或网络问题，程序会自动回退 akshare，不影响使用。")
|
|
print("=" * 60)
|