615 lines
29 KiB
Python
615 lines
29 KiB
Python
"""
|
||
signal_engine.py — V5 短线交易信号引擎（PostgreSQL版）
|
||
|
||
架构：
|
||
- 独立PM2进程，每5秒循环
|
||
- 内存滚动窗口计算指标（CVD/ATR/VWAP/大单阈值）
|
||
- 启动时回灌历史数据（冷启动warmup）
|
||
- 信号评估：核心3条件+加分3条件
|
||
- 输出：signal_indicators表 + signal_trades表 + Discord推送
|
||
|
||
指标：
|
||
- CVD_fast (30m滚动) / CVD_mid (4h滚动) / CVD_day (UTC日内)
|
||
- ATR (5m, 14周期)
|
||
- VWAP_30m
|
||
- 大单阈值 P95/P99 (24h滚动)
|
||
"""
|
||
|
||
import logging
|
||
import os
|
||
import time
|
||
import threading
|
||
import asyncio
|
||
from datetime import datetime, timezone
|
||
from typing import Optional
|
||
|
||
import json
|
||
import websockets
|
||
|
||
from db import get_sync_conn, init_schema
|
||
from signal_state import SymbolState as BaseSymbolState
|
||
from strategy_scoring import evaluate_signal as score_strategy
|
||
from strategy_loader import (
|
||
load_strategy_configs,
|
||
load_strategy_configs_from_db,
|
||
)
|
||
from paper_trading import (
|
||
paper_open_trade,
|
||
paper_has_active_position,
|
||
paper_get_active_direction,
|
||
paper_close_by_signal,
|
||
paper_active_count,
|
||
)
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
handlers=[
|
||
logging.StreamHandler(),
|
||
logging.FileHandler(os.path.join(os.path.dirname(__file__), "..", "signal-engine.log")),
|
||
],
|
||
)
|
||
logger = logging.getLogger("signal-engine")
|
||
|
||
SYMBOLS = ["BTCUSDT", "ETHUSDT", "XRPUSDT", "SOLUSDT"]
|
||
LOOP_INTERVAL = 15 # 秒（从5改15，CPU降60%，信号质量无影响）
|
||
STRATEGY_DIR = os.path.join(os.path.dirname(__file__), "strategies")
|
||
|
||
# ─── 模拟盘配置 ───────────────────────────────────────────────────
|
||
PAPER_TRADING_ENABLED = False # 总开关（兼容旧逻辑）
|
||
PAPER_ENABLED_STRATEGIES = [] # 分策略开关: ["v51_baseline", "v52_8signals"]
|
||
PAPER_INITIAL_BALANCE = 10000 # 虚拟初始资金 USDT
|
||
PAPER_RISK_PER_TRADE = 0.02 # 单笔风险 2%（即200U）
|
||
PAPER_MAX_POSITIONS = 4 # 每套策略最大同时持仓数
|
||
PAPER_TIER_MULTIPLIER = { # 档位仓位倍数
|
||
"light": 0.5, # 轻仓: 1%
|
||
"standard": 1.0, # 标准: 2%
|
||
"heavy": 1.5, # 加仓: 3%
|
||
}
|
||
PAPER_FEE_RATE = 0.0005 # Taker手续费 0.05%（开仓+平仓各一次）
|
||
|
||
def load_paper_config():
|
||
"""从配置文件加载模拟盘开关和参数"""
|
||
global PAPER_TRADING_ENABLED, PAPER_ENABLED_STRATEGIES, PAPER_INITIAL_BALANCE, PAPER_RISK_PER_TRADE, PAPER_MAX_POSITIONS
|
||
config_path = os.path.join(os.path.dirname(__file__), "paper_config.json")
|
||
try:
|
||
with open(config_path, "r") as f:
|
||
import json as _json2
|
||
cfg = _json2.load(f)
|
||
PAPER_TRADING_ENABLED = cfg.get("enabled", False)
|
||
PAPER_ENABLED_STRATEGIES = cfg.get("enabled_strategies", [])
|
||
PAPER_INITIAL_BALANCE = cfg.get("initial_balance", 10000)
|
||
PAPER_RISK_PER_TRADE = cfg.get("risk_per_trade", 0.02)
|
||
PAPER_MAX_POSITIONS = cfg.get("max_positions", 4)
|
||
except FileNotFoundError:
|
||
pass
|
||
|
||
|
||
def is_strategy_enabled(strategy_name: str) -> bool:
|
||
"""检查某策略是否启用模拟盘"""
|
||
if not PAPER_TRADING_ENABLED:
|
||
return False
|
||
# 如果enabled_strategies为空，走旧逻辑（全部启用）
|
||
if not PAPER_ENABLED_STRATEGIES:
|
||
return True
|
||
return strategy_name in PAPER_ENABLED_STRATEGIES
|
||
# ─────────────────────────────────────────────────────────────────
|
||
|
||
# 窗口大小（毫秒）
|
||
WINDOW_FAST = 30 * 60 * 1000 # 30分钟
|
||
WINDOW_MID = 4 * 3600 * 1000 # 4小时
|
||
WINDOW_DAY = 24 * 3600 * 1000 # 24小时
|
||
WINDOW_VWAP = 30 * 60 * 1000 # 30分钟
|
||
|
||
# ATR参数
|
||
ATR_PERIOD_MS = 5 * 60 * 1000
|
||
ATR_LENGTH = 14
|
||
|
||
|
||
def fetch_market_indicators(symbol: str) -> dict:
|
||
"""从PG读取最新的market_indicators数据，解析JSONB提取关键数值"""
|
||
import json as _json
|
||
with get_sync_conn() as conn:
|
||
with conn.cursor() as cur:
|
||
indicators = {}
|
||
ind_types = [
|
||
"long_short_ratio", "top_trader_position", "open_interest_hist",
|
||
"coinbase_premium", "funding_rate",
|
||
"obi_depth_10", "spot_perp_divergence", "tiered_cvd_whale", # Phase 2
|
||
]
|
||
for ind_type in ind_types:
|
||
cur.execute(
|
||
"SELECT value FROM market_indicators WHERE symbol=%s AND indicator_type=%s ORDER BY timestamp_ms DESC LIMIT 1",
|
||
(symbol, ind_type),
|
||
)
|
||
row = cur.fetchone()
|
||
if not row or row[0] is None:
|
||
indicators[ind_type] = None
|
||
continue
|
||
val = row[0]
|
||
if isinstance(val, str):
|
||
try:
|
||
val = _json.loads(val)
|
||
except Exception:
|
||
indicators[ind_type] = None
|
||
continue
|
||
# 提取关键数值
|
||
if ind_type == "long_short_ratio":
|
||
indicators[ind_type] = float(val.get("longShortRatio", 1.0))
|
||
elif ind_type == "top_trader_position":
|
||
indicators[ind_type] = float(val.get("longAccount", 0.5))
|
||
elif ind_type == "open_interest_hist":
|
||
indicators[ind_type] = float(val.get("sumOpenInterestValue", 0))
|
||
elif ind_type == "coinbase_premium":
|
||
indicators[ind_type] = float(val.get("premium_pct", 0)) / 100.0
|
||
elif ind_type == "funding_rate":
|
||
indicators[ind_type] = float(val.get("fundingRate", val.get("lastFundingRate", 0)))
|
||
elif ind_type == "obi_depth_10":
|
||
# obi范围[-1,1]，正值=买压，负值=卖压
|
||
indicators[ind_type] = float(val.get("obi", 0))
|
||
elif ind_type == "spot_perp_divergence":
|
||
# divergence = (spot - mark) / mark
|
||
indicators[ind_type] = float(val.get("divergence", 0))
|
||
elif ind_type == "tiered_cvd_whale":
|
||
# 巨鲸净CVD比率[-1,1]，正值=净买入
|
||
indicators[ind_type] = float(val.get("whale_cvd_ratio", 0))
|
||
return indicators
|
||
|
||
|
||
class SymbolState(BaseSymbolState):
|
||
"""
|
||
兼容旧接口的 SymbolState 包装：
|
||
- 旧代码只传入 symbol，这里补全窗口与 ATR 配置。
|
||
- 评分逻辑统一在 strategy_scoring.evaluate_signal 中实现。
|
||
"""
|
||
|
||
def __init__(self, symbol: str):
|
||
super().__init__(
|
||
symbol,
|
||
WINDOW_FAST,
|
||
WINDOW_MID,
|
||
WINDOW_DAY,
|
||
WINDOW_VWAP,
|
||
ATR_PERIOD_MS,
|
||
ATR_LENGTH,
|
||
fetch_market_indicators,
|
||
)
|
||
|
||
# 评分相关方法已抽离到 backend/strategy_scoring.py，
|
||
# 这里保留一个纯状态容器封装，不再实现打分逻辑。
|
||
|
||
# ─── PG DB操作 ───────────────────────────────────────────────────
|
||
# ─── PG DB操作 ───────────────────────────────────────────────────
|
||
|
||
def load_historical(state: SymbolState, window_ms: int):
|
||
now_ms = int(time.time() * 1000)
|
||
start_ms = now_ms - window_ms
|
||
count = 0
|
||
with get_sync_conn() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT agg_id, price, qty, time_ms, is_buyer_maker FROM agg_trades "
|
||
"WHERE symbol = %s AND time_ms >= %s ORDER BY agg_id ASC",
|
||
(state.symbol, start_ms)
|
||
)
|
||
while True:
|
||
rows = cur.fetchmany(5000)
|
||
if not rows:
|
||
break
|
||
for r in rows:
|
||
state.process_trade(r[0], r[3], r[1], r[2], r[4])
|
||
count += 1
|
||
logger.info(f"[{state.symbol}] 冷启动完成: 加载{count:,}条历史数据 (窗口={window_ms//3600000}h)")
|
||
state.warmup = False
|
||
|
||
|
||
def fetch_new_trades(symbol: str, last_id: int) -> list:
|
||
with get_sync_conn() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT agg_id, price, qty, time_ms, is_buyer_maker FROM agg_trades "
|
||
"WHERE symbol = %s AND agg_id > %s ORDER BY agg_id ASC LIMIT 10000",
|
||
(symbol, last_id)
|
||
)
|
||
return [{"agg_id": r[0], "price": r[1], "qty": r[2], "time_ms": r[3], "is_buyer_maker": r[4]}
|
||
for r in cur.fetchall()]
|
||
|
||
|
||
def save_indicator(ts: int, symbol: str, result: dict, strategy: str = "v52_8signals",
|
||
strategy_id: Optional[str] = None, strategy_name_snapshot: Optional[str] = None):
|
||
with get_sync_conn() as conn:
|
||
with conn.cursor() as cur:
|
||
import json as _json3
|
||
factors_json = _json3.dumps(result.get("factors")) if result.get("factors") else None
|
||
cvd_fast_5m = result.get("cvd_fast_5m") # v53_fast 专用：5m窗口CVD，其他策略为None
|
||
cur.execute(
|
||
"INSERT INTO signal_indicators "
|
||
"(ts,symbol,strategy,cvd_fast,cvd_mid,cvd_day,cvd_fast_slope,atr_5m,atr_percentile,atr_value,vwap_30m,price,p95_qty,p99_qty,score,signal,factors,cvd_fast_5m,strategy_id,strategy_name_snapshot) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(ts, symbol, strategy, result["cvd_fast"], result["cvd_mid"], result["cvd_day"], result["cvd_fast_slope"],
|
||
result["atr"], result["atr_pct"], result.get("atr_value", result["atr"]),
|
||
result["vwap"], result["price"],
|
||
result["p95"], result["p99"], result["score"], result.get("signal"), factors_json,
|
||
cvd_fast_5m, strategy_id, strategy_name_snapshot)
|
||
)
|
||
# 有信号时通知live_executor
|
||
if result.get("signal"):
|
||
cur.execute("NOTIFY new_signal, %s", (f"{symbol}:{strategy}:{result['signal']}:{result['score']}",))
|
||
conn.commit()
|
||
|
||
|
||
def save_feature_event(ts: int, symbol: str, result: dict, strategy: str):
|
||
"""
|
||
V5.3 专用：每次评分后把 raw features + score 层写入 signal_feature_events。
|
||
只对 v53_alt / v53_btc 调用，其他策略跳过。
|
||
"""
|
||
if not strategy.startswith("v53"):
|
||
return
|
||
f = result.get("factors") or {}
|
||
track = f.get("track", "ALT")
|
||
side = result.get("direction") or ("LONG" if result.get("score", 0) >= 0 else "SHORT")
|
||
score_direction = (f.get("direction") or {}).get("score", 0) if track == "ALT" else (f.get("direction") or {}).get("score", 0)
|
||
score_crowding = (f.get("crowding") or {}).get("score", 0)
|
||
score_env = (f.get("environment") or {}).get("score", 0)
|
||
score_aux = (f.get("auxiliary") or {}).get("score", 0)
|
||
gate_passed = f.get("gate_passed", True)
|
||
block_reason = f.get("gate_block") or f.get("block_reason")
|
||
|
||
with get_sync_conn() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO signal_feature_events
|
||
(ts, symbol, track, side, strategy, strategy_version,
|
||
cvd_fast_raw, cvd_mid_raw, cvd_day_raw, cvd_fast_slope_raw,
|
||
p95_qty_raw, p99_qty_raw,
|
||
atr_value, atr_percentile,
|
||
oi_delta_raw, ls_ratio_raw, top_pos_raw, coinbase_premium_raw,
|
||
obi_raw, tiered_cvd_whale_raw,
|
||
score_direction, score_crowding, score_environment, score_aux, score_total,
|
||
gate_passed, block_reason, price)
|
||
VALUES
|
||
(%s,%s,%s,%s,%s,%s,
|
||
%s,%s,%s,%s,
|
||
%s,%s,
|
||
%s,%s,
|
||
%s,%s,%s,%s,
|
||
%s,%s,
|
||
%s,%s,%s,%s,%s,
|
||
%s,%s,%s)
|
||
""",
|
||
(
|
||
ts, symbol, track, side, strategy, "v5.3",
|
||
result.get("cvd_fast"), result.get("cvd_mid"), result.get("cvd_day"), result.get("cvd_fast_slope"),
|
||
result.get("p95"), result.get("p99"),
|
||
result.get("atr_value", result.get("atr")), result.get("atr_pct"),
|
||
result.get("oi_delta"), result.get("ls_ratio"), result.get("top_trader_position"),
|
||
(f.get("auxiliary") or {}).get("coinbase_premium"),
|
||
f.get("obi_raw"), f.get("whale_cvd_ratio"),
|
||
score_direction, score_crowding, score_env, score_aux, result.get("score", 0),
|
||
gate_passed, block_reason, result.get("price"),
|
||
)
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def save_indicator_1m(ts: int, symbol: str, result: dict):
|
||
bar_ts = (ts // 60000) * 60000
|
||
with get_sync_conn() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute("SELECT id FROM signal_indicators_1m WHERE ts=%s AND symbol=%s", (bar_ts, symbol))
|
||
if cur.fetchone():
|
||
cur.execute(
|
||
"UPDATE signal_indicators_1m SET cvd_fast=%s,cvd_mid=%s,cvd_day=%s,atr_5m=%s,vwap_30m=%s,price=%s,score=%s,signal=%s WHERE ts=%s AND symbol=%s",
|
||
(result["cvd_fast"], result["cvd_mid"], result["cvd_day"], result["atr"], result["vwap"],
|
||
result["price"], result["score"], result.get("signal"), bar_ts, symbol)
|
||
)
|
||
else:
|
||
cur.execute(
|
||
"INSERT INTO signal_indicators_1m (ts,symbol,cvd_fast,cvd_mid,cvd_day,atr_5m,vwap_30m,price,score,signal) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(bar_ts, symbol, result["cvd_fast"], result["cvd_mid"], result["cvd_day"], result["atr"],
|
||
result["vwap"], result["price"], result["score"], result.get("signal"))
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
# ─── 模拟盘 ──────────────────────────────────────────────────────
|
||
|
||
def strategy_applies_to_symbol(strategy_cfg: dict, symbol: str) -> bool:
|
||
"""
|
||
判断策略是否适用于当前 symbol。
|
||
- V5.4 单币策略：strategy_cfg.symbol 必须匹配；
|
||
- 兼容旧版多币策略：若配置了 symbols，则需在列表内。
|
||
"""
|
||
strategy_symbol = strategy_cfg.get("symbol")
|
||
if strategy_symbol and strategy_symbol != symbol:
|
||
return False
|
||
allowed_symbols = strategy_cfg.get("symbols", [])
|
||
if allowed_symbols and symbol not in allowed_symbols:
|
||
return False
|
||
return True
|
||
|
||
|
||
|
||
# ─── 实时 WebSocket 数据（OBI + 期现背离）────────────────────────
|
||
|
||
_REALTIME_STATES: dict = {} # symbol -> SymbolState，在main()里注入
|
||
|
||
async def _ws_obi_stream(symbol: str, state):
|
||
"""订阅期货盘口深度流，实时更新 state.rt_obi"""
|
||
stream = f"{symbol.lower()}@depth10@100ms"
|
||
url = f"wss://fstream.binance.com/stream?streams={stream}"
|
||
while True:
|
||
try:
|
||
async with websockets.connect(url, ping_interval=20, ping_timeout=10) as ws:
|
||
logger.info(f"[RT-OBI] {symbol} WebSocket connected")
|
||
async for raw in ws:
|
||
data = json.loads(raw)
|
||
book = data.get("data", data)
|
||
bids = book.get("b", [])
|
||
asks = book.get("a", [])
|
||
bid_vol = sum(float(b[1]) for b in bids)
|
||
ask_vol = sum(float(a[1]) for a in asks)
|
||
total = bid_vol + ask_vol
|
||
state.rt_obi = (bid_vol - ask_vol) / total if total > 0 else 0.0
|
||
except Exception as e:
|
||
logger.warning(f"[RT-OBI] {symbol} 断线重连: {e}")
|
||
await asyncio.sleep(3)
|
||
|
||
async def _ws_spot_perp_stream(symbol: str, state):
|
||
"""
|
||
订阅现货 bookTicker（最优买卖价）+ 期货 markPrice，计算期现背离
|
||
spot_mid = (best_bid + best_ask) / 2
|
||
divergence = (spot_mid - mark_price) / mark_price
|
||
"""
|
||
spot_stream = f"{symbol.lower()}@bookTicker"
|
||
perp_stream = f"{symbol.lower()}@markPrice@1s"
|
||
spot_url = f"wss://stream.binance.com:9443/stream?streams={spot_stream}"
|
||
perp_url = f"wss://fstream.binance.com/stream?streams={perp_stream}"
|
||
|
||
spot_mid = [0.0] # mutable container for closure
|
||
mark_price = [0.0]
|
||
|
||
async def read_spot():
|
||
while True:
|
||
try:
|
||
async with websockets.connect(spot_url, ping_interval=20, ping_timeout=10) as ws:
|
||
logger.info(f"[RT-SPD] {symbol} spot WebSocket connected")
|
||
async for raw in ws:
|
||
d = json.loads(raw).get("data", json.loads(raw))
|
||
bid = float(d.get("b", 0) or 0)
|
||
ask = float(d.get("a", 0) or 0)
|
||
if bid > 0 and ask > 0:
|
||
spot_mid[0] = (bid + ask) / 2
|
||
if mark_price[0] > 0:
|
||
state.rt_spot_perp_div = (spot_mid[0] - mark_price[0]) / mark_price[0]
|
||
except Exception as e:
|
||
logger.warning(f"[RT-SPD] {symbol} spot 断线: {e}")
|
||
await asyncio.sleep(3)
|
||
|
||
async def read_perp():
|
||
while True:
|
||
try:
|
||
async with websockets.connect(perp_url, ping_interval=20, ping_timeout=10) as ws:
|
||
logger.info(f"[RT-SPD] {symbol} perp WebSocket connected")
|
||
async for raw in ws:
|
||
d = json.loads(raw).get("data", json.loads(raw))
|
||
mp = float(d.get("p", 0) or d.get("markPrice", 0) or 0)
|
||
if mp > 0:
|
||
mark_price[0] = mp
|
||
if spot_mid[0] > 0:
|
||
state.rt_spot_perp_div = (spot_mid[0] - mark_price[0]) / mark_price[0]
|
||
except Exception as e:
|
||
logger.warning(f"[RT-SPD] {symbol} perp 断线: {e}")
|
||
await asyncio.sleep(3)
|
||
|
||
await asyncio.gather(read_spot(), read_perp())
|
||
|
||
async def _realtime_ws_runner(states: dict):
|
||
"""统一启动所有实时WebSocket协程，BTC + ALT (ETH/XRP/SOL)"""
|
||
coros = []
|
||
for sym, state in states.items():
|
||
# OBI流：所有symbol都接（perp depth10）
|
||
coros.append(_ws_obi_stream(sym, state))
|
||
# 期现背离流：只有有现货+合约的币种（BTC/ETH/XRP/SOL都有）
|
||
coros.append(_ws_spot_perp_stream(sym, state))
|
||
if coros:
|
||
await asyncio.gather(*coros)
|
||
|
||
def start_realtime_ws(states: dict):
|
||
"""在独立线程里跑asyncio event loop，驱动实时WebSocket采集"""
|
||
def _run():
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
try:
|
||
loop.run_until_complete(_realtime_ws_runner(states))
|
||
except Exception as e:
|
||
logger.error(f"[RT-WS] event loop 异常: {e}")
|
||
finally:
|
||
loop.close()
|
||
|
||
t = threading.Thread(target=_run, daemon=True, name="realtime-ws")
|
||
t.start()
|
||
logger.info("[RT-WS] 实时WebSocket后台线程已启动（BTC OBI + 期现背离）")
|
||
|
||
|
||
# ─── 主循环 ──────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
init_schema()
|
||
# V5.4: 优先从 DB 加载策略配置，失败 fallback 到 JSON
|
||
strategy_configs = load_strategy_configs_from_db()
|
||
if not strategy_configs:
|
||
logger.warning("[DB] 未能从 DB 加载策略配置，使用 JSON fallback")
|
||
strategy_configs = load_strategy_configs()
|
||
strategy_names = [cfg.get("name", "unknown") for cfg in strategy_configs]
|
||
logger.info(f"已加载策略配置: {', '.join(strategy_names)}")
|
||
# 选取 primary 策略用于 1m 指标表：优先 v53 系列，其次第一个配置
|
||
primary_candidates = ["v53", "v53_middle", "v53_fast"]
|
||
primary_strategy_name = None
|
||
for cand in primary_candidates:
|
||
if any(cfg.get("name") == cand for cfg in strategy_configs):
|
||
primary_strategy_name = cand
|
||
break
|
||
if not primary_strategy_name:
|
||
primary_strategy_name = strategy_names[0] if strategy_names else "v53"
|
||
|
||
states = {sym: SymbolState(sym) for sym in SYMBOLS}
|
||
|
||
for sym, state in states.items():
|
||
load_historical(state, WINDOW_MID)
|
||
|
||
logger.info("=== Signal Engine (PG) 启动完成 ===")
|
||
|
||
# 启动实时WebSocket后台线程（BTC OBI + 期现背离）
|
||
start_realtime_ws(states)
|
||
|
||
last_1m_save = {}
|
||
cycle = 0
|
||
warmup_cycles = 3 # 启动后跳过前3轮（45秒），避免冷启动信号开仓
|
||
|
||
while True:
|
||
try:
|
||
now_ms = int(time.time() * 1000)
|
||
# 每轮重新加载配置（支持API热更新开关）
|
||
load_paper_config()
|
||
for sym, state in states.items():
|
||
new_trades = fetch_new_trades(sym, state.last_processed_id)
|
||
for t in new_trades:
|
||
state.process_trade(t["agg_id"], t["time_ms"], t["price"], t["qty"], t["is_buyer_maker"])
|
||
|
||
state.market_indicators = fetch_market_indicators(sym)
|
||
snapshot = state.build_evaluation_snapshot(now_ms)
|
||
strategy_results: list[tuple[dict, dict]] = []
|
||
for strategy_cfg in strategy_configs:
|
||
if not strategy_applies_to_symbol(strategy_cfg, sym):
|
||
continue
|
||
strategy_result = score_strategy(
|
||
state, now_ms, strategy_cfg=strategy_cfg, snapshot=snapshot
|
||
)
|
||
strategy_results.append((strategy_cfg, strategy_result))
|
||
|
||
if not strategy_results:
|
||
logger.warning(f"[{sym}] 当前无可用策略配置，跳过本轮")
|
||
continue
|
||
|
||
# 每个策略独立存储indicator
|
||
for strategy_cfg, strategy_result in strategy_results:
|
||
sname = strategy_cfg.get("name") or "v53"
|
||
sid = strategy_cfg.get("strategy_id")
|
||
ssnap = strategy_cfg.get("strategy_name_snapshot")
|
||
save_indicator(now_ms, sym, strategy_result, strategy=sname,
|
||
strategy_id=sid, strategy_name_snapshot=ssnap)
|
||
save_feature_event(now_ms, sym, strategy_result, strategy=sname)
|
||
|
||
# 1m表仍用primary（图表用）
|
||
primary_result = strategy_results[0][1]
|
||
for strategy_cfg, strategy_result in strategy_results:
|
||
if strategy_cfg.get("name") == primary_strategy_name:
|
||
primary_result = strategy_result
|
||
break
|
||
|
||
bar_1m = (now_ms // 60000) * 60000
|
||
if last_1m_save.get(sym) != bar_1m:
|
||
save_indicator_1m(now_ms, sym, primary_result)
|
||
last_1m_save[sym] = bar_1m
|
||
|
||
# 反向信号平仓：按策略独立判断，score>=每策略配置的 flip_threshold 才触发
|
||
if warmup_cycles <= 0:
|
||
for strategy_cfg, result in strategy_results:
|
||
strategy_name = strategy_cfg.get("name") or "v53"
|
||
if not is_strategy_enabled(strategy_name):
|
||
continue
|
||
# V5.4: custom策略只处理自己配置的symbol
|
||
# per-strategy 方向约束：只接受与策略方向一致的反向信号
|
||
dir_cfg_raw = (strategy_cfg.get("direction") or "both").lower()
|
||
if dir_cfg_raw not in ("long", "short", "both"):
|
||
dir_cfg_raw = "both"
|
||
eval_dir = result.get("direction")
|
||
existing_dir = paper_get_active_direction(sym, strategy_name)
|
||
# 反向平仓分数门槛：优先使用每个策略自己的 flip_threshold，fallback 到 entry_threshold/75
|
||
flip_thr = int(strategy_cfg.get("flip_threshold", strategy_cfg.get("threshold", 75)))
|
||
# 如果策略配置为只多/只空，则忽略与配置相反方向的评估结果
|
||
if dir_cfg_raw == "long" and eval_dir == "SHORT":
|
||
continue
|
||
if dir_cfg_raw == "short" and eval_dir == "LONG":
|
||
continue
|
||
if existing_dir and eval_dir and existing_dir != eval_dir and result["score"] >= flip_thr:
|
||
paper_close_by_signal(sym, result["price"], now_ms, strategy_name)
|
||
logger.info(
|
||
f"[{sym}] 📝 反向信号平仓[{strategy_name}]: {existing_dir} → {eval_dir} "
|
||
f"(score={result['score']}, flip_thr={flip_thr})"
|
||
)
|
||
|
||
for strategy_cfg, result in strategy_results:
|
||
strategy_name = strategy_cfg.get("name") or "v53"
|
||
if result.get("signal"):
|
||
logger.info(
|
||
f"[{sym}] 🚨 信号[{strategy_name}]: {result['signal']} "
|
||
f"score={result['score']} price={result['price']:.1f}"
|
||
)
|
||
# 模拟盘开仓（需该策略启用 + 跳过冷启动）
|
||
if is_strategy_enabled(strategy_name) and warmup_cycles <= 0:
|
||
if not paper_has_active_position(sym, strategy_name):
|
||
active_count = paper_active_count(strategy_name)
|
||
if active_count < PAPER_MAX_POSITIONS:
|
||
tier = result.get("tier", "standard")
|
||
paper_open_trade(
|
||
sym,
|
||
result["signal"],
|
||
result["price"],
|
||
result["score"],
|
||
tier,
|
||
result["atr"],
|
||
now_ms,
|
||
factors=result.get("factors"),
|
||
strategy=strategy_name,
|
||
tp_sl=strategy_cfg.get("tp_sl"),
|
||
strategy_id=strategy_cfg.get("strategy_id"),
|
||
strategy_name_snapshot=strategy_cfg.get("strategy_name_snapshot"),
|
||
)
|
||
|
||
# 模拟盘持仓检查由paper_monitor.py通过WebSocket实时处理，这里不再检查
|
||
|
||
cycle += 1
|
||
if warmup_cycles > 0:
|
||
warmup_cycles -= 1
|
||
if warmup_cycles == 0:
|
||
logger.info("冷启动保护期结束，模拟盘开仓已启用")
|
||
# 每10轮(约2-3分钟)热加载配置，不需要重启
|
||
if cycle % 10 == 0:
|
||
old_strategies = list(PAPER_ENABLED_STRATEGIES)
|
||
load_paper_config()
|
||
# V5.4: 热重载优先读 DB，失败 fallback 到 JSON
|
||
new_configs = load_strategy_configs_from_db()
|
||
if new_configs:
|
||
strategy_configs = new_configs
|
||
else:
|
||
strategy_configs = load_strategy_configs() # A1: 热重载权重/阈值/TP/SL
|
||
strategy_names = [cfg.get("name", "unknown") for cfg in strategy_configs]
|
||
primary_candidates = ["v53", "v53_middle", "v53_fast"]
|
||
primary_strategy_name = None
|
||
for cand in primary_candidates:
|
||
if any(cfg.get("name") == cand for cfg in strategy_configs):
|
||
primary_strategy_name = cand
|
||
break
|
||
if not primary_strategy_name:
|
||
primary_strategy_name = strategy_names[0] if strategy_names else "v53"
|
||
if list(PAPER_ENABLED_STRATEGIES) != old_strategies:
|
||
logger.info(f"📋 配置热加载: enabled_strategies={PAPER_ENABLED_STRATEGIES}")
|
||
for sym, state in states.items():
|
||
logger.info(
|
||
f"[{sym}] 状态: CVD_fast={state.win_fast.cvd:.1f} "
|
||
f"CVD_mid={state.win_mid.cvd:.1f} ATR={state.atr_calc.atr:.2f} "
|
||
f"({state.atr_calc.atr_percentile:.0f}%) VWAP={state.win_vwap.vwap:.1f}"
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"循环异常: {e}", exc_info=True)
|
||
|
||
time.sleep(LOOP_INTERVAL)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|