def fetch_with_fallback(code, start, end):
"""多数据源容错获取"""
sources = ['tdx', 'tushare', 'ths']
for source in sources:
try:
data = QA.QA_fetch_get_stock_day(
package=source,
code=code,
start=start,
end=end
)
if data is not None and len(data) > 0:
print(f"✅ 使用数据源: {source}")
return data
except Exception as e:
print(f"❌ {source} 失败: {e}")
continue
return None
# 使用
data = fetch_with_fallback('000001', '2024-01-01', '2024-12-31')
import QUANTAXIS as QA
from datetime import datetime, timedelta
def update_stock_data(code_list, days=30):
"""更新最近N天的股票数据"""
end_date = datetime.now().strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
for code in code_list:
try:
# 从通达信获取最新数据
data = QA.QA_fetch_get_stock_day(
package='tdx',
code=code,
start=start_date,
end=end_date
)
# 保存到MongoDB
QA.QA_SU_save_stock_day(data)
print(f"✅ {code} 更新成功")
except Exception as e:
print(f"❌ {code} 更新失败: {e}")
# 使用
stock_list = ['000001', '000002', '600000']
update_stock_data(stock_list)
def batch_fetch_stock_day(code_list, start, end, batch_size=50):
"""批量获取股票数据"""
results = {}
for i in range(0, len(code_list), batch_size):
batch = code_list[i:i+batch_size]
for code in batch:
try:
data = QA.QA_fetch_stock_day(
code=code,
start=start,
end=end
)
results[code] = data
except Exception as e:
print(f"❌ {code}: {e}")
print(f"✅ 已完成 {min(i+batch_size, len(code_list))}/{len(code_list)}")
return results
# 使用
codes = QA.QA_fetch_stock_list()['code'].tolist()
data_dict = batch_fetch_stock_day(codes[:100], '2024-01-01', '2024-12-31')
def validate_data(data):
"""验证数据完整性"""
if data is None or len(data) == 0:
return False, "数据为空"
# 检查必要字段
required_cols = ['code', 'date', 'open', 'high', 'low', 'close', 'volume']
missing = [col for col in required_cols if col not in data.columns]
if missing:
return False, f"缺少字段: {missing}"
# 检查空值
null_cols = data.columns[data.isnull().any()].tolist()
if null_cols:
return False, f"存在空值: {null_cols}"
# 检查异常值
if (data['high'] < data['low']).any():
return False, "存在high < low的异常数据"
if (data['close'] > data['high']).any() or (data['close'] < data['low']).any():
return False, "存在close超出high/low范围的异常数据"
return True, "数据验证通过"
# 使用
data = QA.QA_fetch_stock_day('000001', '2024-01-01', '2024-12-31')
is_valid, message = validate_data(data)
print(message)
import pickle
from pathlib import Path
class DataCache:
"""数据缓存管理"""
def __init__(self, cache_dir='./cache'):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
def get_cache_path(self, key):
return self.cache_dir / f"{key}.pkl"
def save(self, key, data):
"""保存数据到缓存"""
cache_path = self.get_cache_path(key)
with open(cache_path, 'wb') as f:
pickle.dump(data, f)
def load(self, key):
"""从缓存加载数据"""
cache_path = self.get_cache_path(key)
if not cache_path.exists():
return None
with open(cache_path, 'rb') as f:
return pickle.load(f)
def is_valid(self, key, max_age_hours=24):
"""检查缓存是否有效"""
cache_path = self.get_cache_path(key)
if not cache_path.exists():
return False
from datetime import datetime
file_time = datetime.fromtimestamp(cache_path.stat().st_mtime)
age_hours = (datetime.now() - file_time).total_seconds() / 3600
return age_hours < max_age_hours
# 使用
cache = DataCache()
def fetch_with_cache(code, start, end):
"""带缓存的数据获取"""
cache_key = f"stock_{code}_{start}_{end}"
# 尝试从缓存加载
if cache.is_valid(cache_key):
data = cache.load(cache_key)
if data is not None:
print(f"✅ 从缓存加载: {cache_key}")
return data
# 从数据库获取
data = QA.QA_fetch_stock_day(code, start, end)
# 保存到缓存
cache.save(cache_key, data)
print(f"✅ 已缓存: {cache_key}")
return data
# 测试
data = fetch_with_cache('000001', '2024-01-01', '2024-12-31')
def robust_fetch(code, start, end):
# 优先从MongoDB查询
try:
data = QA.QA_fetch_stock_day(code, start, end)
if data is not None and len(data) > 0:
return data
except:
pass
# 从通达信获取
try:
data = QA.QA_fetch_get_stock_day('tdx', code, start, end)
if data is not None:
QA.QA_SU_save_stock_day(data) # 保存到MongoDB
return data
except:
pass
# 从Tushare获取
try:
data = QA.QA_fetch_get_stock_day('tushare', code, start, end)
if data is not None:
QA.QA_SU_save_stock_day(data)
return data
except:
pass
raise Exception(f"所有数据源都无法获取 {code} 的数据")
# 不复权
data = QA.QA_fetch_get_stock_day('tdx', '000001', '2020-01-01', '2024-12-31', if_fq='00')
# 前复权（推荐用于回测）
data = QA.QA_fetch_get_stock_day('tdx', '000001', '2020-01-01', '2024-12-31', if_fq='01')
# 后复权
data = QA.QA_fetch_get_stock_day('tdx', '000001', '2020-01-01', '2024-12-31', if_fq='02')
# 手动复权
import QUANTAXIS as QA
data = QA.QA_fetch_stock_day('000001', '2020-01-01', '2024-12-31')
data_qfq = QA.QA_data_stock_to_qfq(data) # 前复权
data_hfq = QA.QA_data_stock_to_hfq(data) # 后复权
