feat: 全链路新增 type 字段支持 + exchange.ts 超时退出优化
- TS: exit 函数统一管理进程退出与 DB 连接关闭；10s 超时 + 异常路径 clearTimeout - Python: PairType(spot/um/cm) 贯穿 Kline 模型、策略配置、数据查询 - 回测脚本升级: 9策略 × 4币种 × 6时间级别 × 2交易类型 - 新增 generate_report.py 回测报告生成工具
This commit is contained in:
Rekey
+38
-15
@@ -27,7 +27,7 @@ from typing import AsyncGenerator
|
||||
import asyncpg
|
||||
|
||||
from ..common.config import DBConfig
|
||||
from ..common.models import Kline, KlineInterval
|
||||
from ..common.models import Kline, KlineInterval, PairType
|
||||
|
||||
# ── 周期 → 表名映射 ──
|
||||
INTERVAL_TO_TABLE: dict[KlineInterval, str] = {
|
||||
@@ -124,18 +124,25 @@ class DataService:
|
||||
# ── 元数据查询 ──
|
||||
|
||||
async def fetch_available_symbols(
|
||||
self, interval: KlineInterval = "1m"
|
||||
self,
|
||||
interval: KlineInterval = "1m",
|
||||
type: PairType = "spot",
|
||||
exchange: str = "binance",
|
||||
) -> list[str]:
|
||||
"""获取指定周期下所有有数据的交易对"""
|
||||
table = INTERVAL_TO_TABLE[interval]
|
||||
async with self._pool.acquire() as conn:
|
||||
rows = await conn.fetch(
|
||||
f"SELECT DISTINCT symbol FROM {table} ORDER BY symbol"
|
||||
f"SELECT DISTINCT symbol FROM {table} "
|
||||
f"WHERE exchange = $1 AND type = $2 ORDER BY symbol",
|
||||
exchange, type,
|
||||
)
|
||||
return [r["symbol"] for r in rows]
|
||||
|
||||
async def fetch_symbol_date_range(
|
||||
self, symbol: str, interval: KlineInterval
|
||||
self, symbol: str, interval: KlineInterval,
|
||||
type: PairType = "spot",
|
||||
exchange: str = "binance",
|
||||
) -> tuple[datetime, datetime]:
|
||||
"""获取指定交易对 + 周期的数据起止时间"""
|
||||
table = INTERVAL_TO_TABLE[interval]
|
||||
@@ -144,9 +151,9 @@ class DataService:
|
||||
f"""
|
||||
SELECT MIN(time) AS min_time, MAX(time) AS max_time
|
||||
FROM {table}
|
||||
WHERE symbol = $1
|
||||
WHERE exchange = $1 AND symbol = $2 AND type = $3
|
||||
""",
|
||||
symbol,
|
||||
exchange, symbol, type,
|
||||
)
|
||||
if row is None or row["min_time"] is None:
|
||||
raise ValueError(f"无数据: {symbol} {interval}")
|
||||
@@ -158,12 +165,14 @@ class DataService:
|
||||
interval: KlineInterval,
|
||||
start_time: datetime | None = None,
|
||||
end_time: datetime | None = None,
|
||||
type: PairType = "spot",
|
||||
exchange: str = "binance",
|
||||
) -> int:
|
||||
"""获取指定条件的 K 线条数（用于预判数据量）"""
|
||||
table = INTERVAL_TO_TABLE[interval]
|
||||
conditions = ["symbol = $1", "interval = $2"]
|
||||
params: list = [symbol, interval]
|
||||
idx = 3
|
||||
conditions = ["exchange = $1", "symbol = $2", "type = $3"]
|
||||
params: list = [exchange, symbol, type]
|
||||
idx = 4
|
||||
|
||||
if start_time is not None:
|
||||
conditions.append(f"time >= ${idx}")
|
||||
@@ -191,6 +200,8 @@ class DataService:
|
||||
end_time: datetime | None = None,
|
||||
limit: int = 1000,
|
||||
offset: int = 0,
|
||||
type: PairType = "spot",
|
||||
exchange: str = "binance",
|
||||
) -> list[Kline]:
|
||||
"""获取 K 线数据，返回 Pydantic 模型列表
|
||||
|
||||
@@ -201,6 +212,8 @@ class DataService:
|
||||
end_time: 结束时间（不包含）
|
||||
limit: 最大返回条数
|
||||
offset: 分页偏移
|
||||
type: 交易对类型（spot / um / cm），默认 spot
|
||||
exchange: 交易所标识，默认 binance
|
||||
|
||||
Returns:
|
||||
按时间升序排列的 Kline 列表
|
||||
@@ -208,9 +221,9 @@ class DataService:
|
||||
table = INTERVAL_TO_TABLE[interval]
|
||||
interval_ms = INTERVAL_MS[interval]
|
||||
|
||||
conditions = ["symbol = $1", "interval = $2"]
|
||||
params: list = [symbol, interval]
|
||||
idx = 3
|
||||
conditions = ["exchange = $1", "symbol = $2", "type = $3"]
|
||||
params: list = [exchange, symbol, type]
|
||||
idx = 4
|
||||
|
||||
if start_time is not None:
|
||||
conditions.append(f"time >= ${idx}")
|
||||
@@ -225,7 +238,7 @@ class DataService:
|
||||
cols = await self._get_columns(table)
|
||||
|
||||
select_cols = [
|
||||
"time", "exchange", "symbol", "interval",
|
||||
"time", "exchange", "symbol", "type", "interval",
|
||||
"open", "high", "low", "close", "volume",
|
||||
]
|
||||
for extra in (
|
||||
@@ -249,7 +262,7 @@ class DataService:
|
||||
offset,
|
||||
)
|
||||
|
||||
return [self._row_to_kline(r, interval, interval_ms) for r in rows]
|
||||
return [self._row_to_kline(r, interval, interval_ms, type) for r in rows]
|
||||
|
||||
async def stream_klines(
|
||||
self,
|
||||
@@ -258,6 +271,8 @@ class DataService:
|
||||
start_time: datetime | None = None,
|
||||
end_time: datetime | None = None,
|
||||
batch_size: int = DEFAULT_BATCH_SIZE,
|
||||
type: PairType = "spot",
|
||||
exchange: str = "binance",
|
||||
) -> AsyncGenerator[list[Kline], None]:
|
||||
"""流式获取 K 线数据，适合大数据集
|
||||
|
||||
@@ -275,6 +290,8 @@ class DataService:
|
||||
end_time=end_time,
|
||||
limit=batch_size,
|
||||
offset=offset,
|
||||
type=type,
|
||||
exchange=exchange,
|
||||
)
|
||||
if not batch:
|
||||
break
|
||||
@@ -288,6 +305,8 @@ class DataService:
|
||||
start_time: datetime | None = None,
|
||||
end_time: datetime | None = None,
|
||||
limit: int = 1000,
|
||||
type: PairType = "spot",
|
||||
exchange: str = "binance",
|
||||
) -> dict[str, list[Kline]]:
|
||||
"""批量获取多个交易对的 K 线
|
||||
|
||||
@@ -301,6 +320,8 @@ class DataService:
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
limit=limit,
|
||||
type=type,
|
||||
exchange=exchange,
|
||||
)
|
||||
for sym in symbols
|
||||
]
|
||||
@@ -326,7 +347,8 @@ class DataService:
|
||||
|
||||
@staticmethod
|
||||
def _row_to_kline(
|
||||
row: asyncpg.Record, interval: KlineInterval, interval_ms: int
|
||||
row: asyncpg.Record, interval: KlineInterval, interval_ms: int,
|
||||
pair_type: PairType = "spot",
|
||||
) -> Kline:
|
||||
"""将数据库行转换为 Kline 模型"""
|
||||
open_time = dt_to_unix_ms(row["time"])
|
||||
@@ -335,6 +357,7 @@ class DataService:
|
||||
exchange=row["exchange"],
|
||||
symbol=row["symbol"],
|
||||
interval=interval,
|
||||
type=row.get("type", pair_type),
|
||||
openTime=open_time,
|
||||
closeTime=open_time + interval_ms,
|
||||
open=_to_float(row["open"]),
|
||||
|
||||
Reference in New Issue
Block a user