feat: 为 FlaskAPIDataSource 添加交易日历获取功能

- 新增 get_trading_calendar() 方法,支持 A股/美股/港股
- 新增 get_calendar_info() 方法,获取服务信息
- 支持自动重试、超时保护、详细错误提示
- 返回标准 DatetimeIndex 格式
- 添加端到端测试验证所有市场
This commit is contained in:
2026-05-24 12:26:35 +08:00
parent e2050e319d
commit d07fb8de6d
2 changed files with 198 additions and 0 deletions

View File

@@ -370,6 +370,109 @@ class FlaskAPIDataSource:
except Exception as e:
return {'status': 'error', 'message': str(e), 'available': False}
def get_calendar_info(self) -> Dict:
"""获取交易日历服务信息"""
url = f"{self.base_url}/api/v1/calendar/info"
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
return response.json()
else:
return {"error": f"HTTP {response.status_code}"}
except Exception as e:
return {"error": str(e)}
def get_trading_calendar(
self,
market: str,
start_date: str,
end_date: str
) -> Optional[pd.DatetimeIndex]:
"""
获取交易日历
Args:
market: 市场代码
- 'A''china': A股上交所/深交所,交易日历一致)
- 'US''us': 美股NYSE
- 'HK''hk': 港股HKEX
start_date: 开始日期 YYYY-MM-DD
end_date: 结束日期 YYYY-MM-DD
Returns:
DatetimeIndex: 交易日日期序列,失败返回 None
示例:
# 获取 A 股 2024 年 1 月交易日历
dates = source.get_trading_calendar('A', '2024-01-01', '2024-01-31')
# 获取美股交易日历
dates = source.get_trading_calendar('US', '2024-01-01', '2024-01-15')
"""
url = f"{self.base_url}/api/v1/trading-calendar"
params = {
'market': market,
'start': start_date,
'end': end_date
}
for attempt in range(self.retries):
try:
response = requests.get(
url,
params=params,
timeout=self.timeout
)
if response.status_code != 200:
if attempt < self.retries - 1:
print(f"⚠ 交易日历请求失败 (HTTP {response.status_code}),重试 {attempt + 2}/{self.retries}")
continue
print(f"✗ 交易日历请求失败: HTTP {response.status_code} - {response.text[:100]}")
return None
data = response.json()
# 检查错误
if 'error' in data:
print(f"✗ 交易日历获取失败: {data['error']}")
return None
# 解析交易日期
trading_dates = data.get('trading_dates', [])
if not trading_dates:
print(f"⚠ 市场 {market}{start_date} ~ {end_date} 期间无交易日")
return pd.DatetimeIndex([])
# 转换为 DatetimeIndex
dates = pd.DatetimeIndex(trading_dates)
count = data.get('count', len(dates))
exchange = data.get('exchange', '')
print(f"{market} ({exchange}): {count} 个交易日 ({start_date} ~ {end_date})")
return dates
except requests.exceptions.Timeout:
if attempt < self.retries - 1:
print(f"⚠ 交易日历请求超时,重试 {attempt + 2}/{self.retries}")
continue
print(f"✗ 交易日历请求超时")
return None
except requests.exceptions.RequestException as e:
if attempt < self.retries - 1:
continue
print(f"✗ 交易日历请求异常: {e}")
return None
except (json.JSONDecodeError, requests.exceptions.JSONDecodeError) as e:
print(f"✗ 交易日历 JSON 解析失败: {e}")
return None
return None
def get_service_info(self) -> Dict:
"""获取服务信息"""
url = f"{self.base_url}/"