refactor: ECR-002 拆分 runtime 包并加深 analyze 契约（已审）
将 web/services/runtime.py 拆为 runtime/ 子模块并保持门面兼容；补齐 ESS 文档、门面/契约/TF_DF 测试与 CODE_REVIEW Approve。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
@@ -0,0 +1,321 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import pandas as pd
|
||||
import requests
|
||||
|
||||
from config import DATA_SERVICE_URL
|
||||
from . import state
|
||||
from .state import DEFAULT_SYMBOLS, DEFAULT_TIMEFRAME_LABELS
|
||||
from .timeframes import build_timeframe_labels
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def _parse_time_input(value):
|
||||
if value in (None, '', 0):
|
||||
return None
|
||||
try:
|
||||
return int(float(value))
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def refresh_data_service_metadata(force=False):
|
||||
"""刷新数据服务提供的交易对与周期元信息。"""
|
||||
now = time.time()
|
||||
if not force and state.DATA_SERVICE_AVAILABLE and now - state.SERVICE_METADATA_LAST_REFRESH < 60:
|
||||
return True
|
||||
try:
|
||||
resp = requests.get(f"{DATA_SERVICE_URL}/health", timeout=5)
|
||||
resp.raise_for_status()
|
||||
payload = resp.json()
|
||||
service_symbols = payload.get("symbols") or payload.get("symbol_list") or []
|
||||
base_timeframes = payload.get("timeframes") or payload.get("base_timeframes") or []
|
||||
derived = payload.get("derived_timeframes") or []
|
||||
service_timeframes = list(base_timeframes)
|
||||
for tf in derived:
|
||||
if tf not in service_timeframes:
|
||||
service_timeframes.append(tf)
|
||||
if service_symbols:
|
||||
state.SYMBOLS[:] = service_symbols
|
||||
if service_timeframes:
|
||||
state.TIMEFRAMES.clear()
|
||||
state.TIMEFRAMES.update(build_timeframe_labels(service_timeframes))
|
||||
state.DATA_SERVICE_AVAILABLE = True
|
||||
state.SERVICE_METADATA_LAST_REFRESH = now
|
||||
return True
|
||||
except Exception as exc:
|
||||
logger.warning("无法加载数据服务元信息: %s", exc)
|
||||
if not state.DATA_SERVICE_AVAILABLE:
|
||||
state.TIMEFRAMES.clear()
|
||||
state.TIMEFRAMES.update(DEFAULT_TIMEFRAME_LABELS)
|
||||
state.SYMBOLS[:] = DEFAULT_SYMBOLS
|
||||
state.DATA_SERVICE_AVAILABLE = False
|
||||
return False
|
||||
|
||||
|
||||
def _fetch_kl_from_datasvc(symbol, timeframe, start_ms=None, end_ms=None, limit=None):
|
||||
params = {"symbol": symbol, "tf": timeframe}
|
||||
if start_ms is not None:
|
||||
params["start"] = int(start_ms)
|
||||
if end_ms is not None:
|
||||
params["end"] = int(end_ms)
|
||||
if limit is not None:
|
||||
params["limit"] = limit
|
||||
resp = requests.get(f"{DATA_SERVICE_URL}/api/candles", params=params, timeout=10)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
if not data:
|
||||
return None
|
||||
df = pd.DataFrame(data)
|
||||
if df.empty or "timestamp" not in df.columns:
|
||||
return None
|
||||
numeric_cols = ["open", "high", "low", "close", "volume"]
|
||||
df["timestamp"] = pd.to_numeric(df["timestamp"], errors="coerce")
|
||||
df = df.dropna(subset=["timestamp"])
|
||||
df["timestamp"] = df["timestamp"].astype("int64")
|
||||
for col in numeric_cols:
|
||||
if col in df.columns:
|
||||
df[col] = pd.to_numeric(df[col], errors="coerce")
|
||||
df = df.dropna(subset=numeric_cols)
|
||||
df = df.sort_values("timestamp")
|
||||
if limit and len(df) > limit:
|
||||
df = df.tail(limit)
|
||||
df = df.reset_index(drop=True)
|
||||
df["date"] = pd.to_datetime(df["timestamp"], unit='ms', utc=True).dt.tz_convert('Asia/Shanghai')
|
||||
return df
|
||||
|
||||
|
||||
# 模块加载时尝试预取一次元信息，但失败不阻塞后续流程
|
||||
refresh_data_service_metadata(force=True)
|
||||
|
||||
# A股热门股票
|
||||
# 模板中 A 股下拉仅放默认一项；用户切换到「A股」时由前端请求 /api/a_stocks 填充全市场（约 5500+）
|
||||
A_STOCK_SYMBOLS = [{'symbol': '000001', 'name': '平安银行'}]
|
||||
|
||||
def detect_symbol_type(symbol):
|
||||
"""检测交易对类型：crypto 或 a_stock"""
|
||||
if '/' in symbol and 'USDT' in symbol:
|
||||
return 'crypto'
|
||||
elif len(symbol) == 6 and symbol.isdigit():
|
||||
return 'a_stock'
|
||||
else:
|
||||
return 'unknown'
|
||||
|
||||
def get_kl_data(symbol, timeframe, limit=100000, start_time=None, end_time=None):
|
||||
"""获取K线数据，支持加密货币和A股"""
|
||||
symbol_type = detect_symbol_type(symbol)
|
||||
|
||||
if symbol_type == 'crypto':
|
||||
return get_crypto_kl_data(symbol, timeframe, limit, start_time, end_time)
|
||||
elif symbol_type == 'a_stock':
|
||||
return get_a_stock_kl_data(symbol, timeframe, limit, start_time, end_time)
|
||||
else:
|
||||
return None
|
||||
|
||||
def _get_crypto_kl_data_via_ccxt(symbol, timeframe, limit=100000, start_time=None, end_time=None):
|
||||
"""获取加密货币K线数据，支持分页加载确保获取指定时间范围内的所有数据"""
|
||||
try:
|
||||
# 初始化参数
|
||||
since = None
|
||||
if start_time:
|
||||
try:
|
||||
since = int(start_time)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# 结束时间处理
|
||||
until = None
|
||||
if end_time:
|
||||
try:
|
||||
until = int(end_time)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# 根据时间周期调整每次请求的数据量
|
||||
batch_size = 1000 # 默认批次大小
|
||||
if timeframe in ['1m', '3m', '5m']:
|
||||
batch_size = 1000 # 分钟级数据减少批次大小
|
||||
elif timeframe in ['15m', '30m', '1h']:
|
||||
batch_size = 1000
|
||||
else:
|
||||
batch_size = 1500 # 日线及以上可以获取更多
|
||||
batch_size = 1500 # 默认批次大小
|
||||
# 初始化存储所有K线数据的列表
|
||||
all_ohlcv = []
|
||||
|
||||
# 初始化当前查询的开始时间
|
||||
current_since = since
|
||||
|
||||
# 添加请求计数和最大限制
|
||||
request_count = 0
|
||||
max_requests = 300 # 最大请求次数，防止无限循环
|
||||
|
||||
# 分页加载数据
|
||||
while request_count < max_requests:
|
||||
request_count += 1
|
||||
|
||||
try:
|
||||
# 获取当前页的数据
|
||||
ohlcv = state.exchange.fetch_ohlcv(symbol, timeframe, since=current_since, limit=batch_size)
|
||||
|
||||
# 如果没有获取到数据，结束循环
|
||||
if not ohlcv or len(ohlcv) == 0:
|
||||
break
|
||||
|
||||
# 将获取到的数据添加到总列表中
|
||||
all_ohlcv.extend(ohlcv)
|
||||
|
||||
# 获取最后一条数据的时间戳
|
||||
last_timestamp = ohlcv[-1][0]
|
||||
|
||||
# 如果已达到结束时间，结束循环
|
||||
if until and last_timestamp >= until:
|
||||
break
|
||||
|
||||
# 如果获取的数据条数小于限制数，说明已经获取完所有数据
|
||||
if len(ohlcv) < batch_size:
|
||||
break
|
||||
|
||||
# 更新下一页的开始时间（加1毫秒避免重复）
|
||||
current_since = last_timestamp + 1
|
||||
|
||||
except Exception as e:
|
||||
# 如果单个批次失败，继续尝试下一个批次
|
||||
if current_since:
|
||||
# 尝试增加时间跳过可能的问题时间点
|
||||
current_since += 60000 # 跳过1分钟
|
||||
else:
|
||||
break
|
||||
|
||||
# 防止API请求过于频繁
|
||||
time.sleep(0.3) # 减少到0.3秒提高效率
|
||||
|
||||
# 数据为空的情况
|
||||
if not all_ohlcv or len(all_ohlcv) == 0:
|
||||
return None
|
||||
|
||||
# 转换为DataFrame
|
||||
df = pd.DataFrame(all_ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
|
||||
df['date'] = pd.to_datetime(df['timestamp'], unit='ms').dt.tz_localize('UTC').dt.tz_convert('Asia/Shanghai')
|
||||
|
||||
# 在客户端进行结束时间过滤
|
||||
if until:
|
||||
df = df[df['timestamp'] <= until]
|
||||
|
||||
# 去除重复数据
|
||||
df = df.drop_duplicates(subset=['timestamp'])
|
||||
|
||||
# 按时间排序
|
||||
df = df.sort_values('timestamp')
|
||||
|
||||
# 限制数据条数的逻辑 - 优先考虑时间范围
|
||||
if start_time and end_time:
|
||||
# 如果指定了明确的时间范围，返回该时间范围内的所有数据
|
||||
if len(df) > 100000: # 防止数据量过大，设置一个合理的上限
|
||||
df = df.tail(100000).reset_index(drop=True)
|
||||
elif limit and len(df) > limit:
|
||||
# 如果没有指定明确时间范围，使用默认的limit限制
|
||||
df = df.tail(limit).reset_index(drop=True)
|
||||
|
||||
# 如果过滤后没有数据，返回None
|
||||
if len(df) == 0:
|
||||
return None
|
||||
return df
|
||||
|
||||
except Exception as e:
|
||||
return None
|
||||
|
||||
|
||||
def get_crypto_kl_data(symbol, timeframe, limit=100000, start_time=None, end_time=None):
|
||||
"""优先通过本地数据服务获取加密货币K线，失败时回退至交易所API。"""
|
||||
start_ms = _parse_time_input(start_time)
|
||||
end_ms = _parse_time_input(end_time)
|
||||
|
||||
refresh_data_service_metadata()
|
||||
if state.DATA_SERVICE_AVAILABLE:
|
||||
try:
|
||||
df = _fetch_kl_from_datasvc(
|
||||
symbol=symbol,
|
||||
timeframe=timeframe,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
limit=limit,
|
||||
)
|
||||
if df is not None and not df.empty:
|
||||
return df
|
||||
except Exception as exc:
|
||||
logger.warning("数据服务请求失败，准备回退至交易所 API：%s", exc)
|
||||
|
||||
return _get_crypto_kl_data_via_ccxt(symbol, timeframe, limit, start_time, end_time)
|
||||
|
||||
|
||||
def get_a_stock_kl_data(symbol, timeframe, limit=100000, start_time=None, end_time=None):
|
||||
"""获取A股K线数据"""
|
||||
try:
|
||||
# 处理时间戳参数转换为日期字符串
|
||||
start_date = None
|
||||
end_date = None
|
||||
|
||||
if start_time:
|
||||
try:
|
||||
# 尝试解析时间戳（毫秒）
|
||||
start_timestamp = int(start_time)
|
||||
start_date = datetime.fromtimestamp(start_timestamp / 1000).strftime('%Y-%m-%d')
|
||||
except (ValueError, TypeError):
|
||||
# 如果不是时间戳，尝试解析datetime-local格式 (YYYY-MM-DDTHH:MM)
|
||||
try:
|
||||
if 'T' in str(start_time):
|
||||
# datetime-local格式：2025-05-19T06:07
|
||||
start_date = str(start_time).split('T')[0] # 只取日期部分
|
||||
else:
|
||||
start_date = str(start_time)
|
||||
except:
|
||||
start_date = start_time
|
||||
|
||||
if end_time:
|
||||
try:
|
||||
# 尝试解析时间戳（毫秒）
|
||||
end_timestamp = int(end_time)
|
||||
end_date = datetime.fromtimestamp(end_timestamp / 1000).strftime('%Y-%m-%d')
|
||||
except (ValueError, TypeError):
|
||||
# 如果不是时间戳，尝试解析datetime-local格式
|
||||
try:
|
||||
if 'T' in str(end_time):
|
||||
# datetime-local格式：2025-05-26T06:07
|
||||
end_date = str(end_time).split('T')[0] # 只取日期部分
|
||||
else:
|
||||
end_date = str(end_time)
|
||||
except:
|
||||
end_date = end_time
|
||||
|
||||
# 如果用户指定了时间范围，优先获取该范围内的所有数据
|
||||
actual_limit = limit
|
||||
if start_date and end_date:
|
||||
actual_limit = None # 不限制数据条数，获取完整时间范围数据
|
||||
|
||||
# 调用A股数据获取器
|
||||
df = state.china_stock.get_kl_data(symbol, timeframe, start_date, end_date, actual_limit)
|
||||
|
||||
if df is None:
|
||||
return None
|
||||
return df
|
||||
|
||||
except Exception as e:
|
||||
return None
|
||||
|
||||
def load_crypto_symbols(limit=200):
|
||||
"""加载常见USDT永续合约交易对，返回列表"""
|
||||
refresh_data_service_metadata()
|
||||
if state.SYMBOLS:
|
||||
return state.SYMBOLS[:limit]
|
||||
try:
|
||||
markets = state.exchange.load_markets()
|
||||
symbols = [s for s in markets.keys() if '/USDT' in s and ':USDT' in s]
|
||||
return symbols[:limit]
|
||||
except Exception:
|
||||
return DEFAULT_SYMBOLS[:limit]
|
||||
|
||||
Reference in New Issue
Block a user