test_tushare_all.py

# -*- coding: utf-8 -*-
"""
测试所有项目需要的 Tushare 接口
"""
import subprocess
import sys

# 安装 tushare
print("Installing tushare...")
subprocess.check_call([sys.executable, "-m", "pip", "install", "tushare", "-q"])
print("Done.\n")

import tushare as ts
import pandas as pd
from datetime import datetime, timedelta
import time

TOKEN = "432d609bd03a0283be53b058912cc563eeb5f24092b53f3dcfd2e358"
ts.set_token(TOKEN)
pro = ts.pro_api()

print("=" * 80)
print("Tushare All Interfaces Test for Project")
print("=" * 80)

test_results = {}

# 定义测试函数
def test_interface(name, test_func):
    print(f"\n{name}:")
    try:
        result = test_func()
        test_results[name] = {'status': 'SUCCESS', 'result': result}
        print(f"   SUCCESS")
        return True
    except Exception as e:
        error_msg = str(e)
        test_results[name] = {'status': 'FAIL', 'error': error_msg}
        print(f"   FAIL: {error_msg[:80]}")
        return False

# 1. 股票列表
test_interface("1. stock_basic", lambda: {
    'count': len(pro.stock_basic(exchange='', list_status='L'))
})

# 等待避免限流
time.sleep(2)

# 2. 日线行情
test_interface("2. daily", lambda: {
    'count': len(pro.daily(ts_code='000001.SZ', start_date='20240301', end_date='20240315')),
    'fields': list(pro.daily(ts_code='000001.SZ', start_date='20240301', end_date='20240315').columns)
})

time.sleep(2)

# 3. 指数日线
test_interface("3. index_daily", lambda: {
    'count': len(pro.index_daily(ts_code='000300.SH', start_date='20240301', end_date='20240315'))
})

time.sleep(2)

# 4. 指数列表
test_interface("4. index_basic", lambda: {
    'count': len(pro.index_basic(market='SSE'))
})

time.sleep(2)

# 5. 资金流向
test_interface("5. moneyflow", lambda: {
    'count': len(pro.moneyflow(ts_code='000001.SZ', start_date='20240301', end_date='20240315'))
})

time.sleep(2)

# 6. 融资融券
test_interface("6. margin_detail", lambda: {
    'count': len(pro.margin_detail(ts_code='000001.SZ', start_date='20240301', end_date='20240315'))
})

time.sleep(2)

# 7. 每日指标
test_interface("7. daily_basic", lambda: {
    'count': len(pro.daily_basic(ts_code='000001.SZ', start_date='20240301', end_date='20240315')),
    'fields': list(pro.daily_basic(ts_code='000001.SZ', start_date='20240301', end_date='20240315').columns)
})

time.sleep(2)

# 8. 复权因子
test_interface("8. adj_factor", lambda: {
    'count': len(pro.adj_factor(ts_code='000001.SZ', start_date='20240301', end_date='20240315'))
})

time.sleep(2)

# 9. 通用行情接口
test_interface("9. pro_bar", lambda: {
    'count': len(ts.pro_bar(ts_code='000001.SZ', adj='qfq', start_date='20240301', end_date='20240315')),
    'fields': list(ts.pro_bar(ts_code='000001.SZ', adj='qfq', start_date='20240301', end_date='20240315').columns)
})

time.sleep(2)

# 10. 财务指标
test_interface("10. fina_indicator", lambda: {
    'count': len(pro.fina_indicator(ts_code='000001.SZ'))
})

time.sleep(2)

# 11. 分红送股
test_interface("11. dividend", lambda: {
    'count': len(pro.dividend(ts_code='000001.SZ'))
})

time.sleep(2)

# 12. 股东人数
test_interface("12. stk_holdernumber", lambda: {
    'count': len(pro.stk_holdernumber(ts_code='000001.SZ'))
})

time.sleep(2)

# 13. 十大流通股东
test_interface("13. top10_floatholders", lambda: {
    'count': len(pro.top10_floatholders(ts_code='000001.SZ'))
})

time.sleep(2)

# 14. 限售股解禁
test_interface("14. share_float", lambda: {
    'count': len(pro.share_float(ann_date='20240315'))
})

time.sleep(2)

# 15. 交易日历
test_interface("15. trade_cal", lambda: {
    'count': len(pro.trade_cal(start_date='20240301', end_date='20240315'))
})

print("\n" + "=" * 80)
print("SUMMARY")
print("=" * 80)

success_count = sum(1 for r in test_results.values() if r.get('status') == 'SUCCESS')
fail_count = len(test_results) - success_count

print(f"\nTotal: {len(test_results)} interfaces")
print(f"SUCCESS: {success_count}")
print(f"FAIL: {fail_count}")

print("\nDetailed Results:")
for name, result in sorted(test_results.items()):
    status = result.get('status', 'UNKNOWN')
    if status == 'SUCCESS':
        res = result.get('result', {})
        count = res.get('count', 'N/A')
        print(f"  [OK] {name}: {count} records")
    else:
        error = result.get('error', 'Unknown')[:60]
        print(f"  [FAIL] {name}: {error}")

print("\n" + "=" * 80)