tushare_cache_system/example_usage.py at main · FengYang/tushare_cache_system · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
"""
使用tushare_cache_system替换tushare的示例
展示如何无缝切换到缓存系统
"""

# 设置环境变量，确保Python使用UTF-8编码处理所有I/O
import os
os.environ['PYTHONIOENCODING'] = 'utf-8'

# 加载.env文件中的环境变量
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    # 如果没有安装python-dotenv，手动加载.env文件
    env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.env')
    if os.path.exists(env_path):
        with open(env_path, 'r') as f:
            for line in f:
                if line.strip() and not line.startswith('#'):
                    key, value = line.strip().split('=', 1)
                    os.environ[key] = value

# 原tushare使用方式
# import tushare as ts
# ts.set_token('your_token')
# pro = ts.pro_api()

# 替换为tushare_cache_system
import sys

# 添加父目录到系统路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# 导入配置读取器
try:
    from config_reader import read_tushare_token
    token = read_tushare_token()
    if not token:
        logger.warning("无法从config.json读取tushare token，请检查配置文件")
        token = 'your_token'  # 使用默认token作为后备
except ImportError:
    logger.warning("无法导入config_reader，使用默认token")
    token = 'your_token'

import tushare_cache_system as ts

# 导入共享日志模块
try:
    from .shared_logger import get_logger
    logger = get_logger(__name__)
except ImportError:
    logger = ts.logger

# 直接使用pro_api函数，与tushare完全一致
logger.info(f"使用token: {token[:10]}..." if token != 'your_token' else "使用默认token，请设置有效的tushare token")
pro = ts.pro_api(token)  # 使用从config.json读取的token

# 检查API是否正确初始化
logger.info(f"API初始化状态: {'成功' if hasattr(pro, 'pro') and pro.pro is not None else '失败'}")
if hasattr(pro, 'pro') and pro.pro is not None:
    logger.info("Tushare Pro API已正确初始化，可以开始获取数据")
else:
    logger.error("Tushare Pro API初始化失败，请检查token是否有效")

    # 尝试手动初始化API
    try:
        import tushare as ts_original
        ts_original.set_token(token)
        pro.pro = ts_original.pro_api()
        logger.info("手动初始化Tushare API成功")
    except Exception as e:
        logger.error(f"手动初始化Tushare API失败: {e}")

# 确保performance_metrics中有所有必需的键
if not hasattr(pro, 'performance_metrics'):
    pro.performance_metrics = {}
required_keys = ['cache_hits', 'cache_misses', 'api_calls', 'source_timing', 'request_counts']
for key in required_keys:
    if key not in pro.performance_metrics:
        if key == 'source_timing' or key == 'request_counts':
            pro.performance_metrics[key] = {}
        else:
            pro.performance_metrics[key] = 0

# 添加延迟，确保预加载任务完成
import time
time.sleep(2)  # 给预加载任务一些时间完成

# 使用方式与tushare完全一致
def example_usage():
    """示例：获取股票基本信息"""

    # 获取股票基本信息 - 与tushare接口完全一致
    stock_basic = pro.stock_basic(exchange='', list_status='L', fields='ts_code,symbol,name,area,industry,list_date')
    logger.info("股票基本信息:")
    print(stock_basic.head())

    # 获取指数权重 - 如果未实现，会自动调用tushare API
    try:
        index_weight = pro.index_weight(index_code='000300.SH', start_date='20230101', end_date='20230131')
        logger.info("\n沪深300指数权重:")
        print(index_weight.head())
    except Exception as e:
        logger.error(f"\n获取指数权重失败: {e}")

    # 获取日线数据 - 如果未实现，会自动调用tushare API
    try:
        daily_data = pro.daily(ts_code='000001.SZ', start_date='20230101', end_date='20230105')
        logger.info("\n平安银行日线数据:")
        print(daily_data.head())
    except Exception as e:
        logger.error(f"\n获取日线数据失败: {e}")

if __name__ == "__main__":
    example_usage()
    logger.info("\n注意：")
    logger.info("1. 请确保已设置有效的tushare token")
    logger.info("2. 已实现的方法会使用缓存，未实现的方法会自动调用tushare API")
    logger.info("3. 所有方法调用方式与tushare完全一致，无需修改现有代码")