2761 lines
124 KiB
Python
2761 lines
124 KiB
Python
"""
|
||
山寨币监控 — 数据库层
|
||
全量记录筛选结果 + 价格跟踪 + 盈亏验证
|
||
"""
|
||
|
||
import sqlite3
|
||
import json
|
||
import os
|
||
import re
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
|
||
from app.config.config_loader import get_meta, get_screener_section, confirm_state_cooldown_hours
|
||
from app.core.opportunity_lifecycle import (
|
||
apply_entry_quality_gate,
|
||
normalize_json_object,
|
||
derive_display_bucket,
|
||
normalize_action_status,
|
||
is_executed_lifecycle,
|
||
)
|
||
from app.core.signal_taxonomy import signal_codes as build_signal_codes, signal_labels as build_signal_labels
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
DB_PATH = os.getenv("ALPHAX_DB_PATH", str(REPO_ROOT / "data" / "altcoin_monitor.db"))
|
||
|
||
|
||
def get_conn():
|
||
# SQLite 并发治理：Web/API/cron 共用同一 DB，连接必须等待写锁而不是立即失败。
|
||
# journal_mode=WAL 只在初始化/首次连接时设置；busy_timeout 让短写事务排队。
|
||
conn = sqlite3.connect(DB_PATH, timeout=30, isolation_level=None)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA busy_timeout=30000")
|
||
conn.execute("PRAGMA journal_mode=WAL")
|
||
conn.execute("PRAGMA synchronous=NORMAL")
|
||
return conn
|
||
|
||
|
||
def init_db():
|
||
conn = get_conn()
|
||
|
||
# 1. 状态跟踪表（原有的）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS coin_state (
|
||
symbol TEXT PRIMARY KEY,
|
||
state TEXT NOT NULL DEFAULT '蓄力',
|
||
score INTEGER DEFAULT 0,
|
||
anomaly_type TEXT DEFAULT '',
|
||
sector TEXT DEFAULT '',
|
||
leader_status TEXT DEFAULT '',
|
||
detected_at TEXT NOT NULL,
|
||
last_alert_time TEXT DEFAULT '',
|
||
last_alert_level TEXT DEFAULT '',
|
||
detail_json TEXT DEFAULT '{}'
|
||
)
|
||
""")
|
||
|
||
# 2. 筛选记录表（每次筛选全量写入）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS screening_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
scan_time TEXT NOT NULL,
|
||
layer TEXT NOT NULL, -- '粗筛'/'细筛'/'确认'
|
||
symbol TEXT NOT NULL,
|
||
state TEXT NOT NULL, -- 蓄力/加速/爆发/过期
|
||
score INTEGER DEFAULT 0,
|
||
price REAL NOT NULL, -- 筛选时价格
|
||
signals TEXT DEFAULT '', -- 信号列表(json array)
|
||
sector TEXT DEFAULT '',
|
||
leader_status TEXT DEFAULT '',
|
||
is_meme INTEGER DEFAULT 0,
|
||
change_24h REAL DEFAULT 0,
|
||
funding_rate REAL DEFAULT 0,
|
||
detail_json TEXT DEFAULT '{}'
|
||
)
|
||
""")
|
||
|
||
# 3. 推荐表（加速/爆发时生成推荐记录，跟踪最终盈亏）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS recommendation (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
symbol TEXT NOT NULL,
|
||
rec_time TEXT NOT NULL, -- 推荐时间
|
||
rec_state TEXT NOT NULL, -- 加速/爆发
|
||
rec_score INTEGER DEFAULT 0,
|
||
entry_price REAL NOT NULL, -- 推荐时价格
|
||
stop_loss REAL DEFAULT 0,
|
||
tp1 REAL DEFAULT 0,
|
||
tp2 REAL DEFAULT 0,
|
||
sector TEXT DEFAULT '',
|
||
signals TEXT DEFAULT '', -- 触发信号(json)
|
||
is_meme INTEGER DEFAULT 0,
|
||
status TEXT DEFAULT 'active', -- active/hit_tp1/hit_tp2/stopped_out/expired
|
||
current_price REAL DEFAULT 0, -- 最新跟踪价格
|
||
max_price REAL DEFAULT 0, -- 推荐后最高价
|
||
min_price REAL DEFAULT 0, -- 推荐后最低价
|
||
pnl_pct REAL DEFAULT 0, -- 当前盈亏%
|
||
max_pnl_pct REAL DEFAULT 0, -- 最大盈亏%
|
||
max_drawdown_pct REAL DEFAULT 0, -- 最大回撤%
|
||
hit_tp1_time TEXT DEFAULT '',
|
||
hit_tp2_time TEXT DEFAULT '',
|
||
stopped_out_time TEXT DEFAULT '',
|
||
expired_time TEXT DEFAULT '',
|
||
last_track_time TEXT DEFAULT '',
|
||
entry_plan_json TEXT DEFAULT '{}',
|
||
action_status TEXT DEFAULT '持有', -- 持有/可即刻买入/等回踩/衰减/止损/止盈1/止盈2/跟踪止盈/反转
|
||
strategy_version TEXT DEFAULT '',
|
||
direction TEXT DEFAULT '中性',
|
||
force_reason TEXT DEFAULT '',
|
||
base_state TEXT DEFAULT '',
|
||
sector_signal_count INTEGER DEFAULT 0,
|
||
market_context_json TEXT DEFAULT '{}',
|
||
derivatives_context_json TEXT DEFAULT '{}',
|
||
sector_context_json TEXT DEFAULT '{}'
|
||
)
|
||
""")
|
||
|
||
# 4. 价格跟踪表（定时快照推荐币的当前价格）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS price_tracking (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
rec_id INTEGER NOT NULL, -- 关联recommendation.id
|
||
symbol TEXT NOT NULL,
|
||
track_time TEXT NOT NULL,
|
||
price REAL NOT NULL,
|
||
pnl_pct REAL DEFAULT 0,
|
||
FOREIGN KEY (rec_id) REFERENCES recommendation(id)
|
||
)
|
||
""")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_price_tracking_rec_id_id ON price_tracking(rec_id, id DESC)")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_price_tracking_rec_time ON price_tracking(rec_id, track_time DESC)")
|
||
|
||
# 4.1 最新价格缓存表（看板/关注池读取现价用，小表；price_tracking 只保留高频流水/审计用途）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS latest_price_cache (
|
||
symbol TEXT PRIMARY KEY,
|
||
price REAL NOT NULL,
|
||
updated_at TEXT NOT NULL,
|
||
source TEXT DEFAULT 'tracker'
|
||
)
|
||
""")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_latest_price_cache_updated_at ON latest_price_cache(updated_at)")
|
||
|
||
# 5. Cron运行日志（每次粗筛/确认/跟踪执行都写一条汇总）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS cron_run_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
job_name TEXT NOT NULL, -- 粗筛/确认/跟踪
|
||
script_name TEXT NOT NULL, -- altcoin_screener.py / altcoin_confirm.py / price_tracker.py
|
||
run_status TEXT NOT NULL, -- success / error
|
||
result_status TEXT DEFAULT '', -- screened / no_candidates / confirmed / tracked / exception ...
|
||
started_at TEXT NOT NULL,
|
||
finished_at TEXT NOT NULL,
|
||
duration_ms INTEGER DEFAULT 0,
|
||
summary_json TEXT DEFAULT '{}',
|
||
error_message TEXT DEFAULT ''
|
||
)
|
||
""")
|
||
|
||
conn.commit()
|
||
|
||
# 迁移：为已有recommendation表添加action_status字段
|
||
try:
|
||
conn.execute("ALTER TABLE recommendation ADD COLUMN action_status TEXT DEFAULT '持有'")
|
||
conn.commit()
|
||
print("DB迁移: recommendation表已添加action_status字段")
|
||
except Exception:
|
||
pass # 字段已存在，忽略
|
||
|
||
# 迁移：为已有recommendation表添加direction字段(多头启动/空头启动/中性)
|
||
try:
|
||
conn.execute("ALTER TABLE recommendation ADD COLUMN direction TEXT DEFAULT '中性'")
|
||
conn.commit()
|
||
print("DB迁移: recommendation表已添加direction字段")
|
||
except Exception:
|
||
pass # 字段已存在，忽略
|
||
|
||
# 迁移：为recommendation补充增强上下文字段
|
||
for sql, message in [
|
||
("ALTER TABLE recommendation ADD COLUMN force_reason TEXT DEFAULT ''", "DB迁移: recommendation表已添加force_reason字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN base_state TEXT DEFAULT ''", "DB迁移: recommendation表已添加base_state字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN sector_signal_count INTEGER DEFAULT 0", "DB迁移: recommendation表已添加sector_signal_count字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN market_context_json TEXT DEFAULT '{}'", "DB迁移: recommendation表已添加market_context_json字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN derivatives_context_json TEXT DEFAULT '{}'", "DB迁移: recommendation表已添加derivatives_context_json字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN sector_context_json TEXT DEFAULT '{}'", "DB迁移: recommendation表已添加sector_context_json字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN lifecycle_state TEXT DEFAULT 'watching'", "DB迁移: recommendation表已添加lifecycle_state字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN display_bucket TEXT DEFAULT 'watch_pool'", "DB迁移: recommendation表已添加display_bucket字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN execution_status TEXT DEFAULT 'observe'", "DB迁移: recommendation表已添加execution_status字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN state_reason TEXT DEFAULT ''", "DB迁移: recommendation表已添加state_reason字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN entry_triggered INTEGER DEFAULT 0", "DB迁移: recommendation表已添加entry_triggered字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN archived_at TEXT DEFAULT ''", "DB迁移: recommendation表已添加archived_at字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN signal_codes_json TEXT DEFAULT '[]'", "DB迁移: recommendation表已添加signal_codes_json字段"),
|
||
("ALTER TABLE recommendation ADD COLUMN signal_labels_json TEXT DEFAULT '[]'", "DB迁移: recommendation表已添加signal_labels_json字段"),
|
||
]:
|
||
try:
|
||
conn.execute(sql)
|
||
conn.commit()
|
||
print(message)
|
||
except Exception:
|
||
pass # 字段已存在，忽略
|
||
try:
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_rec_active_symbol_bucket ON recommendation(symbol, status, display_bucket, id DESC)")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_rec_display_bucket_time ON recommendation(display_bucket, rec_time DESC)")
|
||
conn.commit()
|
||
except Exception:
|
||
pass
|
||
|
||
# 5. 信号绩效表（每个信号类型的命中率统计，用于动态调权）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS signal_performance (
|
||
signal_type TEXT PRIMARY KEY, -- 信号名称如"N倍放量"、"MACD金叉"、"动K起爆"
|
||
category TEXT DEFAULT '', -- 分类: 前瞻/滞后/PA/量价
|
||
total_count INTEGER DEFAULT 0, -- 总出现次数
|
||
hit_count INTEGER DEFAULT 0, -- 命中次数(推荐后48h内涨>5%)
|
||
miss_count INTEGER DEFAULT 0, -- 失手次数(推荐后48h内没涨或跌)
|
||
hit_rate REAL DEFAULT 0, -- 命中率=hit/(hit+miss)
|
||
avg_pnl REAL DEFAULT 0, -- 该信号关联推荐的平均盈亏%
|
||
weight REAL DEFAULT 1.0, -- 当前权重(动态调整)
|
||
last_updated TEXT DEFAULT '' -- 最后更新时间
|
||
)
|
||
""")
|
||
|
||
# 6. 复盘记录表（每条推荐的复盘归因）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS review_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
rec_id INTEGER NOT NULL, -- 关联recommendation.id
|
||
symbol TEXT NOT NULL,
|
||
review_time TEXT NOT NULL, -- 复盘时间
|
||
outcome TEXT NOT NULL, -- 爆发/横盘/失败
|
||
pnl_48h REAL DEFAULT 0, -- 48小时盈亏%
|
||
max_pnl_48h REAL DEFAULT 0, -- 48小时最大盈亏%
|
||
triggered_signals TEXT DEFAULT '', -- 当时触发的信号(json)
|
||
hit_signals TEXT DEFAULT '', -- 哪些信号命中了(真正起作用的)
|
||
miss_signals TEXT DEFAULT '', -- 哪些信号是假信号(没起作用)
|
||
lesson TEXT DEFAULT '', -- 复盘教训总结
|
||
FOREIGN KEY (rec_id) REFERENCES recommendation(id)
|
||
)
|
||
""")
|
||
|
||
# 7. 漏选复盘表（没选但后来爆发的币）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS missed_explosions (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
symbol TEXT NOT NULL,
|
||
detect_time TEXT NOT NULL, -- 发现时间
|
||
price_at_detect REAL DEFAULT 0, -- 发现时价格
|
||
price_before REAL DEFAULT 0, -- 爆发前价格
|
||
gain_pct REAL DEFAULT 0, -- 爆发涨幅%
|
||
reason_missed TEXT DEFAULT '', -- 为什么没选(粗筛没过/细筛淘汰/确认没过)
|
||
features_detected TEXT DEFAULT '', -- 爆发时有什么特征(json)
|
||
lesson TEXT DEFAULT '' -- 漏选教训
|
||
)
|
||
""")
|
||
|
||
# 8. 策略迭代日志（每天复盘/优化过程显式落库）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS strategy_iteration_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
run_date TEXT NOT NULL,
|
||
created_at TEXT NOT NULL,
|
||
trigger_source TEXT DEFAULT 'daily_review',
|
||
title TEXT NOT NULL,
|
||
summary TEXT DEFAULT '',
|
||
findings_json TEXT DEFAULT '[]',
|
||
problems_json TEXT DEFAULT '[]',
|
||
actions_json TEXT DEFAULT '[]',
|
||
changed_rules_json TEXT DEFAULT '[]',
|
||
metrics_json TEXT DEFAULT '{}',
|
||
related_symbols_json TEXT DEFAULT '[]',
|
||
config_diff_json TEXT DEFAULT '{}',
|
||
effect_summary_json TEXT DEFAULT '{}',
|
||
pollution_summary_json TEXT DEFAULT '{}',
|
||
strategy_version TEXT DEFAULT '',
|
||
version_change_summary TEXT DEFAULT ''
|
||
)
|
||
""")
|
||
|
||
# 8.1 候选规则表：研究结论先入候选，不直接发布到主版本
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS strategy_rule_candidate (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
created_at TEXT NOT NULL,
|
||
source TEXT DEFAULT '',
|
||
rule_type TEXT DEFAULT '',
|
||
signal_name TEXT DEFAULT '',
|
||
rule_description TEXT DEFAULT '',
|
||
support_count INTEGER DEFAULT 0,
|
||
success_count INTEGER DEFAULT 0,
|
||
fail_count INTEGER DEFAULT 0,
|
||
avg_pnl REAL DEFAULT 0,
|
||
max_gain REAL DEFAULT 0,
|
||
max_drawdown REAL DEFAULT 0,
|
||
confidence_score REAL DEFAULT 0,
|
||
sample_size INTEGER DEFAULT 0,
|
||
status TEXT DEFAULT 'candidate',
|
||
release_version TEXT DEFAULT '',
|
||
notes TEXT DEFAULT '',
|
||
source_ref TEXT DEFAULT ''
|
||
)
|
||
""")
|
||
|
||
# 8.2 失败模式表：专门记录失败样本与原因分类
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS strategy_failure_pattern (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
created_at TEXT NOT NULL,
|
||
symbol TEXT NOT NULL,
|
||
version TEXT DEFAULT '',
|
||
failure_type TEXT DEFAULT '',
|
||
failure_reason TEXT DEFAULT '',
|
||
signal_combo TEXT DEFAULT '[]',
|
||
market_context_json TEXT DEFAULT '{}',
|
||
entry_quality_issue TEXT DEFAULT '',
|
||
pnl_pct REAL DEFAULT 0,
|
||
max_drawdown_pct REAL DEFAULT 0,
|
||
lesson TEXT DEFAULT ''
|
||
)
|
||
""")
|
||
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS push_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
symbol TEXT NOT NULL,
|
||
push_type TEXT NOT NULL,
|
||
action_status TEXT DEFAULT '',
|
||
pushed_at TEXT NOT NULL
|
||
)
|
||
""")
|
||
_push_cols = [row[1] for row in conn.execute("PRAGMA table_info(push_log)").fetchall()]
|
||
if "rec_id" not in _push_cols:
|
||
conn.execute("ALTER TABLE push_log ADD COLUMN rec_id INTEGER DEFAULT 0")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_push_lookup ON push_log(symbol, push_type, pushed_at)")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_push_log_rec_action ON push_log(rec_id, push_type, action_status, pushed_at)")
|
||
|
||
# strategy_iteration_log 表字段补齐迁移
|
||
iter_cols = [r[1] for r in conn.execute("PRAGMA table_info(strategy_iteration_log)").fetchall()]
|
||
if iter_cols:
|
||
if "config_diff_json" not in iter_cols:
|
||
conn.execute("ALTER TABLE strategy_iteration_log ADD COLUMN config_diff_json TEXT DEFAULT '{}' ")
|
||
if "effect_summary_json" not in iter_cols:
|
||
conn.execute("ALTER TABLE strategy_iteration_log ADD COLUMN effect_summary_json TEXT DEFAULT '{}' ")
|
||
if "pollution_summary_json" not in iter_cols:
|
||
conn.execute("ALTER TABLE strategy_iteration_log ADD COLUMN pollution_summary_json TEXT DEFAULT '{}' ")
|
||
if "strategy_version" not in iter_cols:
|
||
conn.execute("ALTER TABLE strategy_iteration_log ADD COLUMN strategy_version TEXT DEFAULT '' ")
|
||
if "version_change_summary" not in iter_cols:
|
||
conn.execute("ALTER TABLE strategy_iteration_log ADD COLUMN version_change_summary TEXT DEFAULT '' ")
|
||
for col, default in [
|
||
("success_analysis_json", "'{}'"),
|
||
("failure_analysis_json", "'{}'"),
|
||
("candidate_rules_json", "'[]'"),
|
||
("release_decision", "''"),
|
||
("release_reason", "''"),
|
||
("confidence_level", "''"),
|
||
("promotion_state", "'research_only'"),
|
||
]:
|
||
if col not in iter_cols:
|
||
conn.execute(f"ALTER TABLE strategy_iteration_log ADD COLUMN {col} TEXT DEFAULT {default} ")
|
||
|
||
cand_cols = [row["name"] for row in conn.execute("PRAGMA table_info(strategy_rule_candidate)").fetchall()]
|
||
if cand_cols and "source_ref" not in cand_cols:
|
||
conn.execute("ALTER TABLE strategy_rule_candidate ADD COLUMN source_ref TEXT DEFAULT '' ")
|
||
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_rule_candidate_status ON strategy_rule_candidate(status, created_at)")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_failure_pattern_type ON strategy_failure_pattern(failure_type, created_at)")
|
||
|
||
# 9. 舆情事件表（sentiment_monitor 采集，供PA共振加权）
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS sentiment_events (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
symbol TEXT NOT NULL,
|
||
name TEXT DEFAULT '',
|
||
source TEXT NOT NULL,
|
||
event_type TEXT NOT NULL,
|
||
trend_rank INTEGER DEFAULT 0,
|
||
trend_score INTEGER DEFAULT 0,
|
||
market_cap_rank INTEGER DEFAULT 0,
|
||
extra_json TEXT DEFAULT '{}',
|
||
detected_at TEXT NOT NULL
|
||
)
|
||
""")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_sentiment_lookup ON sentiment_events(symbol, source, detected_at)")
|
||
|
||
conn.commit()
|
||
print("DB初始化完成(9+1表)")
|
||
|
||
|
||
# === 推送去重 ===
|
||
PUSH_COOLDOWN_HOURS = 12
|
||
|
||
|
||
def should_push(symbol: str, push_type: str, action_status: str = "") -> bool:
|
||
"""检查是否可以推送：
|
||
- 如果提供了 action_status：同状态+同type冷却期内已推过 → False，状态变了无条件放行
|
||
- 如果未提供 action_status（如burst）：同type冷却期内任何推过 → False
|
||
"""
|
||
conn = get_conn()
|
||
cutoff = (datetime.now() - timedelta(hours=PUSH_COOLDOWN_HOURS)).isoformat()
|
||
if action_status:
|
||
# 状态感知：只拦同状态重复推，状态变了永远放行
|
||
row = conn.execute(
|
||
"SELECT action_status FROM push_log WHERE symbol=? AND push_type=? AND pushed_at > ? ORDER BY id DESC LIMIT 1",
|
||
(symbol, push_type, cutoff),
|
||
).fetchone()
|
||
conn.close()
|
||
if row is None:
|
||
return True # 冷却期内没推过，放行
|
||
last_status = row[0]
|
||
return last_status != action_status # 状态变了放行，相同则冷却
|
||
else:
|
||
# 无状态（burst）：同type任何推过就冷却
|
||
row = conn.execute(
|
||
"SELECT id FROM push_log WHERE symbol=? AND push_type=? AND pushed_at > ? ORDER BY id DESC LIMIT 1",
|
||
(symbol, push_type, cutoff),
|
||
).fetchone()
|
||
conn.close()
|
||
return row is None
|
||
|
||
|
||
def log_push(symbol: str, push_type: str, action_status: str = "", rec_id: int = 0):
|
||
"""记录一次推送。rec_id 可选，作为主链路推荐记录的可追溯来源。"""
|
||
conn = get_conn()
|
||
try:
|
||
cols = [row[1] for row in conn.execute("PRAGMA table_info(push_log)").fetchall()]
|
||
if "rec_id" in cols:
|
||
conn.execute(
|
||
"INSERT INTO push_log (symbol, push_type, action_status, rec_id, pushed_at) VALUES (?,?,?,?,?)",
|
||
(symbol, push_type, action_status, int(rec_id or 0), datetime.now().isoformat()),
|
||
)
|
||
else:
|
||
conn.execute(
|
||
"INSERT INTO push_log (symbol, push_type, action_status, pushed_at) VALUES (?,?,?,?)",
|
||
(symbol, push_type, action_status, datetime.now().isoformat()),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def get_recommendation_for_push(rec_id: int):
|
||
"""读取单条推荐并派生网站同口径展示状态，供推送层消费。
|
||
|
||
飞书/其他通知渠道只能消费这个主链路派生结果，不能自行基于事件或 entry_plan 做推荐判断。
|
||
"""
|
||
try:
|
||
rec_id = int(rec_id or 0)
|
||
except Exception:
|
||
rec_id = 0
|
||
if rec_id <= 0:
|
||
return None
|
||
conn = get_conn()
|
||
row = conn.execute("""
|
||
SELECT r.*,
|
||
lpc.price AS latest_cache_price,
|
||
lpc.updated_at AS latest_cache_updated_at
|
||
FROM recommendation r
|
||
LEFT JOIN latest_price_cache lpc ON lpc.symbol = r.symbol
|
||
WHERE r.id=?
|
||
""", (rec_id,)).fetchone()
|
||
conn.close()
|
||
if not row:
|
||
return None
|
||
item = dict(row)
|
||
rec_result, rec_result_label = _classify_recommendation_result(item)
|
||
item["recommendation_result"] = rec_result
|
||
item["recommendation_result_label"] = rec_result_label
|
||
return _derive_execution_fields(item)
|
||
|
||
|
||
# ==================== 筛选记录 ====================
|
||
|
||
def log_screening(layer, symbol, state, score, price, signals,
|
||
sector="", leader_status="", is_meme=0,
|
||
change_24h=0, funding_rate=0, detail=None):
|
||
"""记录一次筛选结果"""
|
||
conn = get_conn()
|
||
conn.execute("""
|
||
INSERT INTO screening_log (scan_time, layer, symbol, state, score, price, signals,
|
||
sector, leader_status, is_meme, change_24h, funding_rate, detail_json)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
datetime.now().isoformat(), layer, symbol, state, score, price,
|
||
json.dumps(signals, ensure_ascii=False) if isinstance(signals, list) else signals,
|
||
sector, leader_status, is_meme, change_24h, funding_rate,
|
||
json.dumps(detail, ensure_ascii=False) if detail else "{}",
|
||
))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
# ==================== 推荐记录 ====================
|
||
|
||
def _state_fields_for_storage(status, action_status, execution_status="", reason=""):
|
||
bucket = derive_display_bucket(status or "active", action_status, execution_status)
|
||
return (
|
||
bucket.get("execution_status", execution_status or "observe"),
|
||
bucket.get("display_bucket", "watch_pool"),
|
||
bucket.get("lifecycle_state", "watching"),
|
||
1 if is_executed_lifecycle(status or "active", action_status, bucket.get("execution_status")) else 0,
|
||
reason or "",
|
||
)
|
||
|
||
def _derive_minimal_state_fields(status, action_status, entry_plan=None):
|
||
action = normalize_action_status(action_status, status)
|
||
if action == "可即刻买入":
|
||
execution_status = "buy_now"
|
||
reason = "主链路确认当前入场窗口"
|
||
elif action == "等回踩":
|
||
execution_status = "wait_pullback"
|
||
reason = "等待回踩触发，未触发前不计推荐收益"
|
||
elif action == "持有":
|
||
execution_status = "holding"
|
||
reason = "已进入持仓跟踪"
|
||
elif action in ("止盈1", "止盈2", "跟踪止盈"):
|
||
execution_status = "completed"
|
||
reason = "利润管理/阶段兑现"
|
||
elif action in ("止损", "衰减", "反转", "放弃", "过期", "归档") or status in ("stopped_out", "expired", "invalid", "archived"):
|
||
execution_status = "invalid"
|
||
reason = "机会失效，归入历史复盘"
|
||
else:
|
||
execution_status = "observe"
|
||
reason = "观察池，未触发入场"
|
||
return _state_fields_for_storage(status, action, execution_status, reason)
|
||
|
||
|
||
def _serialized_signal_payload(signals):
|
||
labels = build_signal_labels(signals if isinstance(signals, list) else _normalize_signals(signals))
|
||
codes = build_signal_codes(labels)
|
||
stored_signals = json.dumps(labels, ensure_ascii=False) if isinstance(signals, list) else signals
|
||
return stored_signals, json.dumps(codes, ensure_ascii=False), json.dumps(labels, ensure_ascii=False)
|
||
|
||
|
||
def create_recommendation(symbol, rec_state, rec_score, entry_price,
|
||
stop_loss=0, tp1=0, tp2=0, sector="",
|
||
signals="", is_meme=0, entry_plan=None, direction="中性",
|
||
force_reason="", base_state="", sector_signal_count=0,
|
||
market_context=None, derivatives_context=None, sector_context=None):
|
||
"""创建推荐记录（加速/爆发时调用）
|
||
direction: 多头启动/空头启动/中性 — 推荐的方向标签
|
||
注意：rec_score 入参为原始分(0~30范围)，落库时转为百分制(0~100)，分母 30。
|
||
"""
|
||
# 原始分 → 百分制（分母 30，天花板 100）
|
||
raw_pct = round(rec_score * 100.0 / 30) if rec_score else 0
|
||
rec_score_pct = min(raw_pct, 100)
|
||
strategy_version = str(get_meta().get("strategy_version") or "").strip()
|
||
now = datetime.now().isoformat()
|
||
conn = get_conn()
|
||
|
||
incoming_action = normalize_action_status((entry_plan or {}).get("entry_action", "观察") if entry_plan else "观察", "active")
|
||
incoming_exec, incoming_bucket, incoming_lifecycle, incoming_triggered, incoming_reason = _derive_minimal_state_fields(
|
||
"active", incoming_action, entry_plan or {}
|
||
)
|
||
stored_signals, signal_codes_json, signal_labels_json = _serialized_signal_payload(signals)
|
||
# 当前状态唯一：同一 symbol 同一时间只允许一条可执行/观察主记录；
|
||
# 但兼容粗筛蓄力→加速/爆发的状态迁移测试：无 entry_plan 的旧粗筛记录仍可新建演化轨迹。
|
||
duplicate_cursor = conn.execute(
|
||
"""
|
||
SELECT * FROM recommendation
|
||
WHERE symbol=? AND status='active' AND COALESCE(display_bucket,'watch_pool') != 'history'
|
||
ORDER BY id DESC LIMIT 1
|
||
""",
|
||
(symbol,),
|
||
)
|
||
duplicate_row = duplicate_cursor.fetchone() if hasattr(duplicate_cursor, "fetchone") else None
|
||
if duplicate_row and (entry_plan or duplicate_row["rec_state"] == rec_state):
|
||
# 同一币种已有当前主记录时更新该记录，不再制造多个 active。
|
||
# 无 entry_plan 的粗筛状态迁移仍允许保留演化轨迹。
|
||
existing_id = duplicate_row["id"] if hasattr(duplicate_row, "keys") else duplicate_row[0]
|
||
existing_score = duplicate_row["rec_score"] or 0
|
||
merged_state = rec_state
|
||
merged_score = max(existing_score, rec_score_pct)
|
||
conn.execute("""
|
||
UPDATE recommendation
|
||
SET rec_state=?, rec_score=?, sector=COALESCE(NULLIF(?, ''), sector),
|
||
signals=?, signal_codes_json=?, signal_labels_json=?, is_meme=?, direction=?, strategy_version=?,
|
||
force_reason=COALESCE(NULLIF(?, ''), force_reason),
|
||
base_state=COALESCE(NULLIF(?, ''), base_state),
|
||
sector_signal_count=MAX(COALESCE(sector_signal_count,0), ?),
|
||
entry_plan_json=CASE WHEN ? != '{}' THEN ? ELSE entry_plan_json END,
|
||
market_context_json=?, derivatives_context_json=?, sector_context_json=?,
|
||
action_status=CASE
|
||
WHEN action_status IN ('止盈1','止盈2','止损','跟踪止盈','衰减','反转') THEN action_status
|
||
ELSE COALESCE(NULLIF(?, ''), action_status)
|
||
END,
|
||
execution_status=?, display_bucket=?, lifecycle_state=?, entry_triggered=?, state_reason=?
|
||
WHERE id=?
|
||
""", (
|
||
merged_state, merged_score, sector,
|
||
stored_signals, signal_codes_json, signal_labels_json,
|
||
is_meme, direction, strategy_version,
|
||
force_reason or "", base_state or "", int(sector_signal_count or 0),
|
||
json.dumps(entry_plan or {}, ensure_ascii=False),
|
||
json.dumps(entry_plan or {}, ensure_ascii=False),
|
||
json.dumps(market_context or {}, ensure_ascii=False),
|
||
json.dumps(derivatives_context or {}, ensure_ascii=False),
|
||
json.dumps(sector_context or {}, ensure_ascii=False),
|
||
incoming_action if entry_plan else "",
|
||
incoming_exec, incoming_bucket, incoming_lifecycle, incoming_triggered, incoming_reason,
|
||
existing_id,
|
||
))
|
||
conn.commit()
|
||
conn.close()
|
||
return existing_id
|
||
|
||
cursor = conn.execute("""
|
||
INSERT INTO recommendation (symbol, rec_time, rec_state, rec_score, entry_price,
|
||
stop_loss, tp1, tp2, sector, signals, signal_codes_json, signal_labels_json, is_meme, direction,
|
||
current_price, max_price, min_price, last_track_time, entry_plan_json,
|
||
force_reason, base_state, sector_signal_count,
|
||
market_context_json, derivatives_context_json, sector_context_json,
|
||
action_status, execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason,
|
||
strategy_version)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
symbol, now, rec_state, rec_score_pct, entry_price,
|
||
stop_loss, tp1, tp2, sector,
|
||
stored_signals, signal_codes_json, signal_labels_json,
|
||
is_meme, direction, entry_price, entry_price, entry_price,
|
||
now,
|
||
json.dumps(entry_plan, ensure_ascii=False) if entry_plan else "{}",
|
||
force_reason or "",
|
||
base_state or "",
|
||
int(sector_signal_count or 0),
|
||
json.dumps(market_context or {}, ensure_ascii=False),
|
||
json.dumps(derivatives_context or {}, ensure_ascii=False),
|
||
json.dumps(sector_context or {}, ensure_ascii=False),
|
||
incoming_action,
|
||
incoming_exec, incoming_bucket, incoming_lifecycle, incoming_triggered, incoming_reason,
|
||
strategy_version,
|
||
))
|
||
rec_id = cursor.lastrowid
|
||
conn.commit()
|
||
conn.close()
|
||
return rec_id
|
||
|
||
|
||
def update_recommendation_tracking(rec_id, current_price):
|
||
"""更新推荐记录的跟踪价格和盈亏。
|
||
|
||
v1.7.9+: TP1 只代表阶段兑现/启动跟踪止盈，不再把记录移出 active；
|
||
这样 TP1 后继续推高的收益会继续计入 current/max_pnl。
|
||
"""
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT entry_price, max_price, min_price, symbol FROM recommendation WHERE id=?", (rec_id,)).fetchone()
|
||
if not row:
|
||
conn.close()
|
||
return
|
||
|
||
entry_price = row["entry_price"]
|
||
old_max = row["max_price"] or entry_price
|
||
old_min = row["min_price"] or entry_price
|
||
|
||
new_max = max(old_max, current_price)
|
||
new_min = min(old_min, current_price)
|
||
pnl_pct = round((current_price / entry_price - 1) * 100, 2)
|
||
max_pnl_pct = round((new_max / entry_price - 1) * 100, 2)
|
||
max_drawdown_pct = round((new_min / entry_price - 1) * 100, 2)
|
||
|
||
status = "active"
|
||
tp1_reached = False
|
||
rec = conn.execute("SELECT stop_loss, tp1, tp2, status, hit_tp1_time FROM recommendation WHERE id=?", (rec_id,)).fetchone()
|
||
if rec and rec["status"] == "active":
|
||
if rec["tp2"] and current_price >= rec["tp2"]:
|
||
status = "hit_tp2"
|
||
elif rec["stop_loss"] and current_price <= rec["stop_loss"]:
|
||
status = "stopped_out"
|
||
elif rec["tp1"] and current_price >= rec["tp1"]:
|
||
status = "hit_tp1"
|
||
tp1_reached = True
|
||
elif rec["tp1"] == 0 and pnl_pct >= 15:
|
||
status = "hit_tp1"
|
||
tp1_reached = True
|
||
|
||
now = datetime.now().isoformat()
|
||
if status != "active":
|
||
action_for_status = {"hit_tp1": "止盈1", "hit_tp2": "止盈2", "stopped_out": "止损"}.get(status, "持有")
|
||
execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason = _derive_minimal_state_fields(status, action_for_status, {})
|
||
conn.execute("""
|
||
UPDATE recommendation SET current_price=?, max_price=?, min_price=?,
|
||
pnl_pct=?, max_pnl_pct=?, max_drawdown_pct=?,
|
||
status=?, action_status=?, execution_status=?, display_bucket=?, lifecycle_state=?, entry_triggered=?, state_reason=?, last_track_time=?,
|
||
hit_tp1_time=CASE WHEN ?='hit_tp1' THEN ? ELSE hit_tp1_time END,
|
||
hit_tp2_time=CASE WHEN ?='hit_tp2' THEN ? ELSE hit_tp2_time END,
|
||
stopped_out_time=CASE WHEN ?='stopped_out' THEN ? ELSE stopped_out_time END
|
||
WHERE id=?
|
||
""", (current_price, new_max, new_min, pnl_pct, max_pnl_pct, max_drawdown_pct,
|
||
status, action_for_status, execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason, now,
|
||
status, now, status, now, status, now, rec_id))
|
||
else:
|
||
conn.execute("""
|
||
UPDATE recommendation SET current_price=?, max_price=?, min_price=?,
|
||
pnl_pct=?, max_pnl_pct=?, max_drawdown_pct=?,
|
||
last_track_time=?,
|
||
hit_tp1_time=CASE WHEN ? THEN COALESCE(NULLIF(hit_tp1_time,''), ?) ELSE hit_tp1_time END
|
||
WHERE id=?
|
||
""", (current_price, new_max, new_min, pnl_pct, max_pnl_pct, max_drawdown_pct, now,
|
||
1 if tp1_reached else 0, now, rec_id))
|
||
|
||
symbol = row["symbol"]
|
||
update_latest_price_cache(symbol, current_price, updated_at=now, source="tracker", conn=conn)
|
||
|
||
conn.execute("""
|
||
INSERT INTO price_tracking (rec_id, symbol, track_time, price, pnl_pct)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
""", (rec_id, symbol, now, current_price, pnl_pct))
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
return {"status": status, "tp1_reached": tp1_reached, "pnl_pct": pnl_pct, "max_pnl_pct": max_pnl_pct, "max_drawdown_pct": max_drawdown_pct}
|
||
|
||
|
||
def expire_old_recommendations(hours=48):
|
||
"""超过48小时的active推荐标记为expired"""
|
||
conn = get_conn()
|
||
cutoff = datetime.now().timestamp() - hours * 3600
|
||
conn.execute("""
|
||
UPDATE recommendation SET status='expired', expired_time=?
|
||
WHERE status='active' AND julianday(?) - julianday(rec_time) > ?
|
||
""", (datetime.now().isoformat(), datetime.now().isoformat(), hours / 24.0))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def _entry_window_policy(entry_price, current_price, rec_time, event_time=None, window_hours=2.0, up_deviation_pct=1.5, down_deviation_pct=1.2):
|
||
"""阶段1：入场窗口可信度规则。
|
||
|
||
- 入场窗口默认有效 2 小时；超过后降级观察。
|
||
- 当前价向上脱离触发价 >1.5%：不追高，降级等回踩。
|
||
- 当前价向下跌破触发价 >1.2%：买点失效，降级观察。
|
||
"""
|
||
event_time = event_time or datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
|
||
try:
|
||
entry_price = float(entry_price or 0)
|
||
current_price = float(current_price or 0)
|
||
except Exception:
|
||
entry_price = 0
|
||
current_price = 0
|
||
deviation_pct = round((current_price / entry_price - 1) * 100, 2) if entry_price and current_price else 0.0
|
||
age_minutes = 0.0
|
||
try:
|
||
start = datetime.fromisoformat(str(rec_time))
|
||
end = datetime.fromisoformat(str(event_time))
|
||
age_minutes = round((end - start).total_seconds() / 60.0, 1)
|
||
except Exception:
|
||
age_minutes = 0.0
|
||
remaining_minutes = round(max(0.0, window_hours * 60.0 - age_minutes), 1)
|
||
result = {
|
||
"status": "active",
|
||
"label": "入场窗口有效",
|
||
"reason": "入场窗口仍在有效期内，价格未明显脱离触发价",
|
||
"age_minutes": age_minutes,
|
||
"remaining_minutes": remaining_minutes,
|
||
"window_hours": window_hours,
|
||
"entry_price": entry_price,
|
||
"current_price": current_price,
|
||
"deviation_pct": deviation_pct,
|
||
"max_up_deviation_pct": up_deviation_pct,
|
||
"max_down_deviation_pct": down_deviation_pct,
|
||
}
|
||
if age_minutes > window_hours * 60.0:
|
||
result.update({
|
||
"status": "expired",
|
||
"label": "窗口已过期",
|
||
"reason": f"入场窗口超过有效期 {window_hours:g} 小时，避免沿用旧信号追入",
|
||
"remaining_minutes": 0.0,
|
||
})
|
||
elif deviation_pct > up_deviation_pct:
|
||
result.update({
|
||
"status": "price_left_up",
|
||
"label": "价格已上脱离",
|
||
"reason": f"当前价较触发价上脱离 {deviation_pct:.2f}%，超过 {up_deviation_pct:g}% 阈值，避免追高",
|
||
})
|
||
elif deviation_pct < -down_deviation_pct:
|
||
result.update({
|
||
"status": "price_left_down",
|
||
"label": "价格已下破",
|
||
"reason": f"当前价较触发价下破 {abs(deviation_pct):.2f}%，买点动能失效，转观察",
|
||
})
|
||
return result
|
||
|
||
|
||
def _risk_suggestion(entry_price, stop_loss, tp1, risk_budget_pct=1.0, max_position_pct=100.0):
|
||
"""把入场价/止损转换成可执行仓位建议。"""
|
||
try:
|
||
entry_price = float(entry_price or 0)
|
||
stop_loss = float(stop_loss or 0)
|
||
tp1 = float(tp1 or 0)
|
||
except Exception:
|
||
entry_price = stop_loss = tp1 = 0
|
||
stop_distance_pct = round(abs(entry_price - stop_loss) / entry_price * 100, 2) if entry_price and stop_loss else 0.0
|
||
suggested_position_pct = round(min(max_position_pct, risk_budget_pct / stop_distance_pct * 100), 2) if stop_distance_pct else 0.0
|
||
tp1_profit_pct = round((tp1 / entry_price - 1) * 100, 2) if entry_price and tp1 else 0.0
|
||
rr = round(tp1_profit_pct / stop_distance_pct, 2) if stop_distance_pct else 0.0
|
||
max_loss_pct = round(suggested_position_pct * stop_distance_pct / 100, 2) if suggested_position_pct else 0.0
|
||
return {
|
||
"risk_budget_pct": risk_budget_pct,
|
||
"stop_distance_pct": stop_distance_pct,
|
||
"suggested_position_pct": suggested_position_pct,
|
||
"max_loss_pct": max_loss_pct,
|
||
"tp1_profit_pct": tp1_profit_pct,
|
||
"rr": rr,
|
||
"max_position_pct": max_position_pct,
|
||
"valid": bool(entry_price and stop_loss and stop_distance_pct > 0),
|
||
}
|
||
|
||
|
||
def update_latest_price_cache(symbol, price, updated_at=None, source="tracker", conn=None):
|
||
"""Upsert 最新行情缓存。看板读取这张小表，不再依赖 price_tracking 高频流水表。"""
|
||
symbol = str(symbol or "").strip().upper()
|
||
try:
|
||
price = float(price or 0)
|
||
except Exception:
|
||
price = 0
|
||
if not symbol or price <= 0:
|
||
return False
|
||
updated_at = updated_at or datetime.now().isoformat()
|
||
owns_conn = conn is None
|
||
if owns_conn:
|
||
conn = get_conn()
|
||
conn.execute("""
|
||
INSERT INTO latest_price_cache (symbol, price, updated_at, source)
|
||
VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(symbol) DO UPDATE SET
|
||
price=excluded.price,
|
||
updated_at=excluded.updated_at,
|
||
source=excluded.source
|
||
""", (symbol, price, updated_at, source))
|
||
if owns_conn:
|
||
conn.commit()
|
||
conn.close()
|
||
return True
|
||
|
||
|
||
def get_latest_price_cache(symbols):
|
||
"""批量读取最新行情缓存，返回 {symbol: {price, updated_at, source}}。"""
|
||
normalized = []
|
||
for sym in symbols or []:
|
||
sym = str(sym or "").strip().upper()
|
||
if sym and sym not in normalized:
|
||
normalized.append(sym)
|
||
if not normalized:
|
||
return {}
|
||
conn = get_conn()
|
||
placeholders = ",".join(["?"] * len(normalized))
|
||
rows = conn.execute(
|
||
f"SELECT symbol, price, updated_at, source FROM latest_price_cache WHERE symbol IN ({placeholders})",
|
||
tuple(normalized),
|
||
).fetchall()
|
||
conn.close()
|
||
return {row["symbol"]: dict(row) for row in rows}
|
||
|
||
|
||
def _latest_tracking_price(rec_id, fallback=0):
|
||
"""旧兼容函数：不再读取 price_tracking，避免看板/API 依赖高频流水大表。
|
||
|
||
最新现价应来自 latest_price_cache；没有缓存时回退 recommendation.current_price。
|
||
"""
|
||
return fallback or 0
|
||
|
||
|
||
def _execution_fields_from_persisted_state(item, entry_plan=None):
|
||
"""只基于DB主状态派生展示状态；不得用 entry_plan.initial_action 反向提升主状态。"""
|
||
entry_plan = entry_plan if entry_plan is not None else _normalize_entry_plan(item.get("entry_plan_json"))
|
||
status = (item.get("status") or "active").strip()
|
||
action_status = normalize_action_status(item.get("action_status") or "持有", status)
|
||
|
||
bucket = derive_display_bucket(status, action_status, "")
|
||
lifecycle = bucket.get("lifecycle_state")
|
||
execution_status = bucket.get("execution_status")
|
||
if execution_status == "completed":
|
||
return "completed", "✅ 已兑现，仅观察", f"该机会已进入{action_status or '利润管理'}阶段，仅作为持仓跟踪记录"
|
||
if execution_status == "invalid":
|
||
if action_status == "止损":
|
||
reason = "该机会已触发风险边界，原入场逻辑失效"
|
||
elif action_status == "衰减":
|
||
reason = "该机会已出现趋势衰减，追高性价比下降"
|
||
elif action_status == "反转":
|
||
reason = "该机会已出现趋势反转，原多头逻辑被破坏"
|
||
elif action_status == "放弃":
|
||
reason = "该机会已被标记为放弃，不再满足入场条件"
|
||
else:
|
||
reason = "该机会观察周期结束或逻辑失效，已归入历史复盘"
|
||
return "invalid", "🔴 已失效，勿追", reason
|
||
if execution_status == "buy_now":
|
||
stop = str(entry_plan.get("stop_loss", "")) if entry_plan else ""
|
||
return "buy_now", "🟢 现在可买", "推荐时就是可即刻买入；主链路确认当前仍在入场窗口" + (("，风险边界 " + stop) if stop else "")
|
||
if execution_status == "wait_pullback":
|
||
gate = entry_plan.get("entry_quality_gate") or {}
|
||
if gate.get("reasons"):
|
||
reason = "等待更优位置；" + "；".join(gate.get("reasons", [])[:3])
|
||
else:
|
||
reason = "等待回踩至 " + (str(entry_plan.get("entry_price", "")) if entry_plan else "参考价") + " 附近再评估"
|
||
return "wait_pullback", "🟡 等回踩，不追高", reason
|
||
if execution_status == "holding":
|
||
return "holding", "持仓跟踪", "该机会已触发入场，进入持仓跟踪"
|
||
gate = entry_plan.get("entry_quality_gate") or {}
|
||
if gate.get("reasons"):
|
||
reason = "机会结构仍在观察；" + "；".join(gate.get("reasons", [])[:3])
|
||
else:
|
||
reason = "暂无明确入场窗口，继续观察"
|
||
return "observe", "观察池", reason
|
||
|
||
|
||
def apply_recommendation_state_transition(rec_id, requested_action, current_price, event_time=None, signals=None):
|
||
"""主链路状态迁移：唯一允许把价格事件落成 action_status 的入口。
|
||
|
||
返回值同时作为推送层 payload 来源；推送层不得再自行判定交易状态。
|
||
"""
|
||
event_time = event_time or datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
|
||
conn = get_conn()
|
||
row = conn.execute("""
|
||
SELECT * FROM recommendation WHERE id=?
|
||
""", (rec_id,)).fetchone()
|
||
if not row:
|
||
conn.close()
|
||
return {"updated": False, "push_required": False, "reason": "not_found"}
|
||
|
||
item = dict(row)
|
||
previous_action = (item.get("action_status") or "持有").strip()
|
||
entry_plan = _normalize_entry_plan(item.get("entry_plan_json"))
|
||
terminal_map = {"hit_tp2": "止盈2", "stopped_out": "止损"}
|
||
status = (item.get("status") or "active").strip()
|
||
final_action = normalize_action_status(terminal_map.get(status, requested_action), status)
|
||
|
||
if status not in terminal_map:
|
||
final_action, entry_plan, gate_reasons = apply_entry_quality_gate(
|
||
action_status=final_action,
|
||
entry_plan=entry_plan,
|
||
signals=signals if signals is not None else item.get("signals"),
|
||
current_price=current_price,
|
||
market_context=normalize_json_object(item.get("market_context_json")),
|
||
derivatives_context=normalize_json_object(item.get("derivatives_context_json")),
|
||
sector_context=normalize_json_object(item.get("sector_context_json")),
|
||
)
|
||
else:
|
||
gate_reasons = []
|
||
|
||
window_entry_price = item.get("entry_price") or current_price or 0
|
||
window_rec_time = item.get("rec_time") or event_time
|
||
if final_action == "可即刻买入" and previous_action != "可即刻买入":
|
||
window_entry_price = current_price
|
||
window_rec_time = event_time
|
||
entry_window = _entry_window_policy(window_entry_price, current_price, window_rec_time, event_time)
|
||
if final_action == "可即刻买入" and previous_action == "可即刻买入":
|
||
if entry_window["status"] == "expired":
|
||
final_action = "观察"
|
||
gate_reasons.append(entry_window["reason"])
|
||
elif entry_window["status"] == "price_left_up":
|
||
final_action = "等回踩"
|
||
gate_reasons.append(entry_window["reason"])
|
||
elif entry_window["status"] == "price_left_down":
|
||
final_action = "观察"
|
||
gate_reasons.append(entry_window["reason"])
|
||
|
||
should_reset_entry = final_action == "可即刻买入" and previous_action != "可即刻买入"
|
||
if should_reset_entry:
|
||
max_price = current_price
|
||
min_price = current_price
|
||
pnl_pct = 0.0
|
||
max_pnl_pct = 0.0
|
||
max_drawdown_pct = 0.0
|
||
rec_time = event_time
|
||
entry_price = current_price
|
||
else:
|
||
old_entry = item.get("entry_price") or current_price or 0
|
||
old_max = item.get("max_price") or old_entry
|
||
old_min = item.get("min_price") or old_entry
|
||
max_price = max(old_max, current_price) if current_price else old_max
|
||
min_price = min(old_min, current_price) if current_price else old_min
|
||
entry_price = old_entry
|
||
rec_time = item.get("rec_time")
|
||
pnl_pct = round((current_price / old_entry - 1) * 100, 2) if old_entry and current_price else item.get("pnl_pct", 0)
|
||
max_pnl_pct = round((max_price / old_entry - 1) * 100, 2) if old_entry else item.get("max_pnl_pct", 0)
|
||
max_drawdown_pct = round((min_price / old_entry - 1) * 100, 2) if old_entry else item.get("max_drawdown_pct", 0)
|
||
|
||
execution_status, execution_label, execution_reason = _execution_fields_from_persisted_state(
|
||
{**item, "action_status": final_action, "status": status}, entry_plan
|
||
)
|
||
execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason = _state_fields_for_storage(
|
||
status, final_action, execution_status, execution_reason
|
||
)
|
||
push_required = final_action in ("可即刻买入", "跟踪止盈") and previous_action != final_action and execution_status in ("buy_now", "completed")
|
||
|
||
conn.execute("""
|
||
UPDATE recommendation
|
||
SET action_status=?, entry_plan_json=?, current_price=?, max_price=?, min_price=?,
|
||
pnl_pct=?, max_pnl_pct=?, max_drawdown_pct=?, last_track_time=?,
|
||
execution_status=?, display_bucket=?, lifecycle_state=?, entry_triggered=?, state_reason=?,
|
||
rec_time=CASE WHEN ? THEN ? ELSE rec_time END,
|
||
entry_price=CASE WHEN ? THEN ? ELSE entry_price END
|
||
WHERE id=?
|
||
""", (
|
||
final_action, json.dumps(entry_plan, ensure_ascii=False), current_price, max_price, min_price,
|
||
pnl_pct, max_pnl_pct, max_drawdown_pct, event_time,
|
||
execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason,
|
||
1 if should_reset_entry else 0, rec_time,
|
||
1 if should_reset_entry else 0, entry_price,
|
||
rec_id,
|
||
))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
return {
|
||
"updated": True,
|
||
"id": rec_id,
|
||
"symbol": item.get("symbol"),
|
||
"previous_action_status": previous_action,
|
||
"action_status": final_action,
|
||
"execution_status": execution_status,
|
||
"execution_label": execution_label,
|
||
"execution_reason": execution_reason,
|
||
"display_bucket": display_bucket,
|
||
"lifecycle_state": lifecycle_state,
|
||
"entry_triggered": entry_triggered,
|
||
"entry_price": entry_price,
|
||
"current_price": current_price,
|
||
"pnl_pct": pnl_pct,
|
||
"stop_loss": item.get("stop_loss") or entry_plan.get("stop_loss") or 0,
|
||
"tp1": item.get("tp1") or entry_plan.get("tp1") or entry_plan.get("take_profit_1") or 0,
|
||
"tp2": item.get("tp2") or entry_plan.get("tp2") or 0,
|
||
"entry_plan": entry_plan,
|
||
"entry_window": entry_window,
|
||
"risk_suggestion": _risk_suggestion(
|
||
entry_price,
|
||
item.get("stop_loss") or entry_plan.get("stop_loss") or 0,
|
||
item.get("tp1") or entry_plan.get("tp1") or entry_plan.get("take_profit_1") or 0,
|
||
),
|
||
"gate_reasons": gate_reasons,
|
||
"push_required": push_required,
|
||
"push_symbol": item.get("symbol"),
|
||
"push_entry_price": entry_price,
|
||
"push_current_price": current_price,
|
||
"push_pnl_pct": pnl_pct,
|
||
"push_signals": signals or [],
|
||
}
|
||
|
||
|
||
|
||
def recompute_all_recommendation_state_fields(conn=None):
|
||
"""回填统一状态机派生字段。只读 status/action_status，不改变历史交易价格。"""
|
||
owns_conn = conn is None
|
||
if owns_conn:
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT id,status,action_status,entry_plan_json FROM recommendation").fetchall()
|
||
updated = 0
|
||
for row in rows:
|
||
ep = _normalize_entry_plan(row["entry_plan_json"])
|
||
action = normalize_action_status(row["action_status"], row["status"])
|
||
execution_status, execution_label, execution_reason = _execution_fields_from_persisted_state(
|
||
{"status": row["status"], "action_status": action, "entry_plan_json": row["entry_plan_json"]}, ep
|
||
)
|
||
execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason = _state_fields_for_storage(
|
||
row["status"], action, execution_status, execution_reason
|
||
)
|
||
conn.execute(
|
||
"""UPDATE recommendation
|
||
SET action_status=?, execution_status=?, display_bucket=?, lifecycle_state=?, entry_triggered=?, state_reason=?
|
||
WHERE id=?""",
|
||
(action, execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason, row["id"]),
|
||
)
|
||
updated += 1
|
||
if owns_conn:
|
||
conn.commit()
|
||
conn.close()
|
||
return updated
|
||
|
||
def update_recommendation_action_status(rec_id, action_status):
|
||
"""更新推荐记录的操作状态。
|
||
|
||
保护规则：如果推荐已经真实止盈/止损结案，不能再被后续动态入场逻辑覆盖成
|
||
“可即刻买入/持有/等回踩”。v1.7.5 进一步增加买点质量闸门：
|
||
entry_plan 为等回踩、risk_reward_ok=false、rr过低或追高距离过远时，拒绝把
|
||
action_status 写成“可即刻买入”。
|
||
"""
|
||
conn = get_conn()
|
||
row = conn.execute("""
|
||
SELECT status, action_status, entry_plan_json, signals, current_price,
|
||
market_context_json, derivatives_context_json, sector_context_json
|
||
FROM recommendation WHERE id=?
|
||
""", (rec_id,)).fetchone()
|
||
terminal_map = {
|
||
"hit_tp1": "止盈1",
|
||
"hit_tp2": "止盈2",
|
||
"stopped_out": "止损",
|
||
}
|
||
entry_plan = {}
|
||
if row:
|
||
if row["status"] in terminal_map and action_status not in ("止盈1", "止盈2", "止损", "跟踪止盈"):
|
||
action_status = terminal_map[row["status"]]
|
||
else:
|
||
entry_plan = _normalize_entry_plan(row["entry_plan_json"])
|
||
gated_action, gated_plan, _ = apply_entry_quality_gate(
|
||
action_status=action_status,
|
||
entry_plan=entry_plan,
|
||
signals=row["signals"],
|
||
current_price=row["current_price"] or 0,
|
||
market_context=normalize_json_object(row["market_context_json"]),
|
||
derivatives_context=normalize_json_object(row["derivatives_context_json"]),
|
||
sector_context=normalize_json_object(row["sector_context_json"]),
|
||
)
|
||
action_status = gated_action
|
||
entry_plan = gated_plan
|
||
if entry_plan:
|
||
execution_status, execution_label, execution_reason = _execution_fields_from_persisted_state(
|
||
{"status": row["status"] if row else "active", "action_status": action_status, "entry_plan_json": json.dumps(entry_plan, ensure_ascii=False)},
|
||
entry_plan,
|
||
)
|
||
execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason = _state_fields_for_storage(
|
||
row["status"] if row else "active", action_status, execution_status, execution_reason
|
||
)
|
||
conn.execute("""
|
||
UPDATE recommendation SET action_status=?, entry_plan_json=?, execution_status=?, display_bucket=?, lifecycle_state=?, entry_triggered=?, state_reason=? WHERE id=?
|
||
""", (action_status, json.dumps(entry_plan, ensure_ascii=False), execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason, rec_id))
|
||
else:
|
||
execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason = _derive_minimal_state_fields(row["status"] if row else "active", action_status, {})
|
||
conn.execute("""
|
||
UPDATE recommendation SET action_status=?, execution_status=?, display_bucket=?, lifecycle_state=?, entry_triggered=?, state_reason=? WHERE id=?
|
||
""", (action_status, execution_status, display_bucket, lifecycle_state, entry_triggered, state_reason, rec_id))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def update_entry_timing(rec_id: int, entry_price: float, rec_time: str):
|
||
"""更新入场到位的时间和价格。当tracker检测到可即刻买入时调用。"""
|
||
conn = get_conn()
|
||
conn.execute(
|
||
"UPDATE recommendation SET rec_time=?, entry_price=?, current_price=?, pnl_pct=0 WHERE id=?",
|
||
(rec_time, entry_price, entry_price, rec_id)
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def _normalize_entry_plan(entry_plan_json):
|
||
try:
|
||
if isinstance(entry_plan_json, dict):
|
||
return entry_plan_json
|
||
if entry_plan_json:
|
||
return json.loads(entry_plan_json)
|
||
except Exception:
|
||
pass
|
||
return {}
|
||
|
||
|
||
def _normalize_json_object(payload):
|
||
try:
|
||
if isinstance(payload, dict):
|
||
return payload
|
||
if payload:
|
||
parsed = json.loads(payload)
|
||
if isinstance(parsed, dict):
|
||
return parsed
|
||
except Exception:
|
||
pass
|
||
return {}
|
||
|
||
|
||
def _normalize_signals(payload):
|
||
"""signals 字段从 SQLite TEXT 列读出是 JSON 字符串，必须解析为数组才能交给前端 JS .map()。"""
|
||
try:
|
||
if isinstance(payload, list):
|
||
return payload
|
||
if isinstance(payload, str) and payload.strip():
|
||
parsed = json.loads(payload)
|
||
if isinstance(parsed, list):
|
||
return parsed
|
||
except Exception:
|
||
pass
|
||
return []
|
||
|
||
|
||
|
||
def _observe_tier(item):
|
||
"""观察池分层：strong=值得用户关注，weak=弱观察/低质量候选。"""
|
||
status = str(item.get("execution_status") or "")
|
||
if status in ("buy_now", "wait_pullback") or item.get("display_bucket") == "realtime":
|
||
return "strong", "入场/等待类有效机会"
|
||
try:
|
||
score = float(item.get("rec_score") or 0)
|
||
except Exception:
|
||
score = 0
|
||
signals = item.get("signals") or []
|
||
if isinstance(signals, str):
|
||
signals = _normalize_signals(signals)
|
||
sig_text = " ".join(str(x) for x in signals)
|
||
force_reason = str(item.get("force_reason") or "")
|
||
derivatives = _normalize_json_object(item.get("derivatives_context_json") or item.get("derivatives_context"))
|
||
market = _normalize_json_object(item.get("market_context_json") or item.get("market_context"))
|
||
if not derivatives and isinstance(item.get("derivatives_context"), dict):
|
||
derivatives = item.get("derivatives_context") or {}
|
||
if not market and isinstance(item.get("market_context"), dict):
|
||
market = item.get("market_context") or {}
|
||
long_pct = 0.0
|
||
try:
|
||
long_pct = float(derivatives.get("top_trader_long_pct") or 0)
|
||
except Exception:
|
||
long_pct = 0.0
|
||
acc1 = 0.0
|
||
acc4 = 0.0
|
||
try:
|
||
acc1 = float(market.get("turnover_acceleration_1h") or 0)
|
||
acc4 = float(market.get("turnover_acceleration_4h") or 0)
|
||
except Exception:
|
||
pass
|
||
stale_only = ("已过期" in sig_text or "历史" in sig_text) and not any(k in sig_text for k in ("当前", "新近", "刚刚", "入场窗口", "量价齐飞"))
|
||
weak_reasons = []
|
||
if score < 50:
|
||
weak_reasons.append(f"评分偏低({int(score)})")
|
||
if stale_only:
|
||
weak_reasons.append("主要触发来自历史/过期信号")
|
||
if "静K蓄力旁路" in force_reason and acc4 < 1.3 and acc1 < 1.3:
|
||
weak_reasons.append("静K旁路量能不足")
|
||
gate = {}
|
||
try:
|
||
ep = item.get("entry_plan") or _normalize_json_object(item.get("entry_plan_json"))
|
||
gate = ep.get("entry_quality_gate") or {}
|
||
except Exception:
|
||
gate = {}
|
||
gate_reasons = gate.get("reasons") or []
|
||
gate_reason_text = "；".join(str(x) for x in gate_reasons[:3])
|
||
if any("回踩参考已到" in str(x) and "不达标" in str(x) for x in gate_reasons):
|
||
return "weak" if score < 55 else "strong", (gate_reason_text or "回踩参考已到，但实时盈亏比不达标") + "；暂不构成入场窗口，继续观察是否重新恢复可买盈亏比"
|
||
strong_context = score >= 65 or long_pct >= 75 or max(acc1, acc4) >= 1.5
|
||
if weak_reasons and not strong_context:
|
||
return "weak", "；".join(weak_reasons[:3])
|
||
if gate_reason_text:
|
||
return "strong", gate_reason_text + "；继续观察结构是否恢复"
|
||
return "strong", "观察池有效候选"
|
||
|
||
def _derive_execution_fields(item):
|
||
entry_plan = _normalize_entry_plan(item.get("entry_plan_json"))
|
||
market_context = _normalize_json_object(item.get("market_context_json"))
|
||
derivatives_context = _normalize_json_object(item.get("derivatives_context_json"))
|
||
sector_context = _normalize_json_object(item.get("sector_context_json"))
|
||
signals = _normalize_signals(item.get("signals"))
|
||
item["signals"] = signals
|
||
initial_action = normalize_action_status(entry_plan.get("entry_action") or item.get("action_status") or "持有", item.get("status") or "active")
|
||
action_status = normalize_action_status(item.get("action_status") or initial_action or "持有", item.get("status") or "active")
|
||
# 新建爆发推荐可能还没被 tracker 跑到，DB action_status 仍是默认“持有”。
|
||
# 此时以前端展示和实时看板过滤应以确认层写入的 entry_plan.entry_action 为准，
|
||
# 但后续 tracker 一旦写入明确状态，仍以 DB 主状态优先。
|
||
if action_status == "持有" and initial_action in ("可即刻买入", "等回踩", "观察"):
|
||
action_status = initial_action
|
||
current_price_for_window = item.get("latest_cache_price") or item.get("current_price") or item.get("entry_price") or 0
|
||
action_status, entry_plan, _entry_gate_reasons = apply_entry_quality_gate(
|
||
action_status=action_status,
|
||
entry_plan=entry_plan,
|
||
signals=item.get("signals"),
|
||
current_price=current_price_for_window,
|
||
market_context=market_context,
|
||
derivatives_context=derivatives_context,
|
||
sector_context=sector_context,
|
||
)
|
||
if initial_action == "可即刻买入" and action_status != "可即刻买入":
|
||
initial_action = action_status
|
||
status = (item.get("status") or "active").strip()
|
||
force_reason = (item.get("force_reason") or "").strip()
|
||
base_state = (item.get("base_state") or "").strip()
|
||
sector_signal_count = item.get("sector_signal_count")
|
||
strategy_version = str(item.get("strategy_version") or "").strip()
|
||
if not strategy_version:
|
||
strategy_version = str(get_meta().get("strategy_version") or "").strip()
|
||
|
||
if current_price_for_window:
|
||
item["current_price"] = current_price_for_window
|
||
try:
|
||
entry_price_for_pnl = float(item.get("entry_price") or 0)
|
||
current_price_float = float(current_price_for_window or 0)
|
||
if entry_price_for_pnl > 0 and current_price_float > 0:
|
||
item["pnl_pct"] = round((current_price_float - entry_price_for_pnl) / entry_price_for_pnl * 100, 2)
|
||
except Exception:
|
||
pass
|
||
if item.get("latest_cache_updated_at"):
|
||
item["current_price_updated_at"] = item.get("latest_cache_updated_at")
|
||
entry_window = _entry_window_policy(
|
||
item.get("entry_price") or entry_plan.get("entry_price") or 0,
|
||
current_price_for_window,
|
||
item.get("rec_time") or "",
|
||
) if action_status == "可即刻买入" else {}
|
||
# 实时看板用 hours 参数过滤过期机会；派生层不再因为旧 rec_time 反向篡改主状态，避免展示/测试口径分裂。
|
||
item_for_execution = {**item, "action_status": action_status}
|
||
execution_status, execution_label, execution_reason = _execution_fields_from_persisted_state(item_for_execution, entry_plan)
|
||
|
||
bucket_fields = derive_display_bucket(status, action_status, execution_status)
|
||
execution_status = bucket_fields.get("execution_status") or execution_status
|
||
item["initial_action"] = initial_action
|
||
item["action_status"] = normalize_action_status(action_status, status)
|
||
item["execution_status"] = execution_status
|
||
item["execution_label"] = execution_label
|
||
item["execution_reason"] = execution_reason
|
||
item["display_bucket"] = bucket_fields.get("display_bucket")
|
||
item["lifecycle_state"] = bucket_fields.get("lifecycle_state")
|
||
item["entry_triggered"] = 1 if is_executed_lifecycle(status, item["action_status"], execution_status) else 0
|
||
# 派生状态可能被买点质量闸门从“等回踩”降为“观察”，同步刷新展示桶，避免卡片仍停留在旧等待态。
|
||
bucket_fields = derive_display_bucket(status, item["action_status"], item["execution_status"])
|
||
item["execution_status"] = bucket_fields.get("execution_status") or item["execution_status"]
|
||
item["display_bucket"] = bucket_fields.get("display_bucket")
|
||
item["lifecycle_state"] = bucket_fields.get("lifecycle_state")
|
||
observe_tier, observe_reason = _observe_tier(item)
|
||
item["observe_tier"] = observe_tier
|
||
item["observe_reason"] = observe_reason
|
||
item["entry_plan"] = entry_plan
|
||
item["entry_window"] = entry_window
|
||
if entry_window and entry_window.get("status") != "active":
|
||
item["entry_window_alert"] = entry_window
|
||
item["risk_suggestion"] = _risk_suggestion(
|
||
item.get("entry_price") or entry_plan.get("entry_price") or 0,
|
||
item.get("stop_loss") or entry_plan.get("stop_loss") or 0,
|
||
item.get("tp1") or entry_plan.get("tp1") or entry_plan.get("take_profit_1") or 0,
|
||
)
|
||
item["market_context"] = market_context
|
||
item["derivatives_context"] = derivatives_context
|
||
item["sector_context"] = sector_context
|
||
item["force_reason"] = force_reason
|
||
item["base_state"] = base_state
|
||
item["sector_signal_count"] = sector_signal_count
|
||
item["strategy_version"] = strategy_version
|
||
item["strategy_version_label"] = f"策略版本 {strategy_version}" if strategy_version else ""
|
||
return item
|
||
|
||
|
||
def _is_actionable_execution_status(status):
|
||
"""实时可操作口径：包含现在可买 + 等回踩计划；但收益只统计已执行交易。"""
|
||
return status in ("buy_now", "wait_pullback")
|
||
|
||
|
||
def _is_executed_trade(item):
|
||
"""收益统计口径：只有真实触发入场/持仓/退出的样本才计算收益。
|
||
|
||
buy_now 是当前入场窗口，不等同于已成交；等回踩/观察永远不计推荐收益。
|
||
"""
|
||
status = (item.get("status") or "").strip()
|
||
action_status = normalize_action_status(item.get("action_status"), status)
|
||
execution_status = item.get("execution_status") or ""
|
||
if action_status == "可即刻买入" or execution_status == "buy_now":
|
||
return True
|
||
return is_executed_lifecycle(status, action_status, execution_status)
|
||
|
||
|
||
def _classify_recommendation_result(item):
|
||
status = item.get("status") or ""
|
||
pnl_pct = item.get("pnl_pct") or 0
|
||
max_pnl_pct = item.get("max_pnl_pct") or 0
|
||
max_drawdown_pct = item.get("max_drawdown_pct") or 0
|
||
|
||
if status in ("hit_tp1", "hit_tp2"):
|
||
return "success", "✅ 止盈成功"
|
||
if status == "stopped_out":
|
||
return "failed", "❌ 止损失败"
|
||
|
||
# 计划/观察未实际触发入场前，不按推荐生成价计算成功/失败。
|
||
if not _is_executed_trade(item):
|
||
return "pending", "⏳ 未执行"
|
||
|
||
if status == "expired":
|
||
if max_pnl_pct >= 5:
|
||
return "success", "✅ 交易成功"
|
||
if pnl_pct <= -3 or max_drawdown_pct <= -5:
|
||
return "failed", "❌ 交易失败"
|
||
return "pending", "⏳ 跟踪中"
|
||
if status == "active":
|
||
if max_pnl_pct >= 5:
|
||
return "success", "✅ 交易成功"
|
||
if pnl_pct <= -3 or max_drawdown_pct <= -5:
|
||
return "failed", "❌ 交易失败"
|
||
return "pending", "⏳ 跟踪中"
|
||
return "pending", "⏳ 未执行"
|
||
|
||
|
||
# ==================== 查询API ====================
|
||
|
||
def get_active_recommendations(actionable_only=False):
|
||
"""获取所有active推荐。默认保留全量，实时页请使用去重视图的可执行口径。"""
|
||
conn = get_conn()
|
||
rows = conn.execute("""
|
||
SELECT * FROM recommendation
|
||
WHERE status='active' AND COALESCE(display_bucket,'watch_pool') != 'history'
|
||
ORDER BY rec_time DESC
|
||
""").fetchall()
|
||
conn.close()
|
||
result = []
|
||
for row in rows:
|
||
item = _derive_execution_fields(dict(row))
|
||
if actionable_only and not _is_actionable_execution_status(item.get("execution_status")):
|
||
continue
|
||
result.append(item)
|
||
return result
|
||
|
||
|
||
def get_active_recommendations_deduped(actionable_only=True, version="", hours=0, watch_symbols=None, limit=0, offset=0, with_meta=False):
|
||
"""获取去重后的active推荐（同symbol只保留最新一条），并附带推荐结果判定。
|
||
version 为空时不按版本过滤；hours>0 时只取最近 N 小时信号。
|
||
with_meta=True 时返回分页对象，兼容实时看板首屏分页加载。"""
|
||
conn = get_conn()
|
||
where = "status='active' AND COALESCE(display_bucket,'watch_pool') != 'history'"
|
||
params = []
|
||
version = str(version or "").strip()
|
||
if version:
|
||
where += " AND strategy_version=?"
|
||
params.append(version)
|
||
if watch_symbols:
|
||
symbols = [str(s).strip().upper() for s in watch_symbols if str(s).strip()]
|
||
if symbols:
|
||
where += " AND symbol IN (" + ",".join(["?"] * len(symbols)) + ")"
|
||
params.extend(symbols)
|
||
try:
|
||
hours = float(hours or 0)
|
||
except Exception:
|
||
hours = 0
|
||
if hours > 0:
|
||
where += " AND julianday(?) - julianday(rec_time) <= ?"
|
||
params.extend([datetime.now().isoformat(), hours / 24.0])
|
||
|
||
try:
|
||
limit = max(0, int(limit or 0))
|
||
except Exception:
|
||
limit = 0
|
||
try:
|
||
offset = max(0, int(offset or 0))
|
||
except Exception:
|
||
offset = 0
|
||
|
||
rows = conn.execute(f"""
|
||
SELECT r.*,
|
||
lpc.price AS latest_cache_price,
|
||
lpc.updated_at AS latest_cache_updated_at
|
||
FROM recommendation r
|
||
LEFT JOIN latest_price_cache lpc ON lpc.symbol = r.symbol
|
||
JOIN (
|
||
SELECT symbol, MAX(id) AS max_id
|
||
FROM recommendation
|
||
WHERE {where}
|
||
GROUP BY symbol
|
||
) latest ON latest.max_id = r.id
|
||
ORDER BY r.rec_time DESC
|
||
""", tuple(params)).fetchall()
|
||
conn.close()
|
||
|
||
all_items = []
|
||
# 实时看板只输出当前有效机会；过期/失效样本属于历史/复盘，不再进入实时列表或 summary。
|
||
summary = {"buy_now": 0, "wait_pullback": 0, "observe": 0, "observe_strong": 0, "observe_weak": 0, "expired": 0, "total": 0}
|
||
now = datetime.now()
|
||
for row in rows:
|
||
item = dict(row)
|
||
rec_result, rec_result_label = _classify_recommendation_result(item)
|
||
item["recommendation_result"] = rec_result
|
||
item["recommendation_result_label"] = rec_result_label
|
||
_derive_execution_fields(item)
|
||
|
||
is_expired = False
|
||
if hours > 0:
|
||
try:
|
||
rec_time = item.get("rec_time")
|
||
if rec_time:
|
||
is_expired = (now - datetime.fromisoformat(str(rec_time))).total_seconds() > hours * 3600
|
||
except Exception:
|
||
is_expired = False
|
||
if item.get("execution_status") == "invalid" or item.get("status") in ("invalid", "expired", "archived") or item.get("display_bucket") == "history":
|
||
is_expired = True
|
||
|
||
# 带 hours 的实时看板请求必须过滤旧/脏/过期；不带 hours 的内部/测试查询保留全量派生结果。
|
||
if is_expired:
|
||
summary["expired"] += 1
|
||
continue
|
||
|
||
if actionable_only and not _is_actionable_execution_status(item.get("execution_status")):
|
||
continue
|
||
all_items.append(item)
|
||
|
||
if item.get("execution_status") == "buy_now":
|
||
summary["buy_now"] += 1
|
||
elif item.get("execution_status") == "wait_pullback":
|
||
summary["wait_pullback"] += 1
|
||
else:
|
||
summary["observe"] += 1
|
||
if item.get("observe_tier") == "weak":
|
||
summary["observe_weak"] += 1
|
||
else:
|
||
summary["observe_strong"] += 1
|
||
summary["total"] = len(all_items)
|
||
# expired 仅作内部审计计数，不属于实时机会流；API 对外不暴露，避免前端/用户继续看到过期入口。
|
||
summary["expired_filtered"] = summary.pop("expired", 0)
|
||
|
||
if not with_meta:
|
||
return all_items
|
||
page_items = all_items[offset: offset + limit] if limit else all_items[offset:]
|
||
return {
|
||
"items": page_items,
|
||
"total": len(all_items),
|
||
"limit": limit,
|
||
"offset": offset,
|
||
"has_more": bool(limit and offset + len(page_items) < len(all_items)),
|
||
"summary": summary,
|
||
}
|
||
|
||
|
||
def get_all_recommendations(limit=50, decision_only=False, version="", offset=0, with_meta=False):
|
||
"""兼容导出：推荐列表查询已迁移到 analytics 模块。"""
|
||
from app.db.analytics import get_all_recommendations as _get_all_recommendations
|
||
|
||
return _get_all_recommendations(
|
||
limit=limit,
|
||
decision_only=decision_only,
|
||
version=version,
|
||
offset=offset,
|
||
with_meta=with_meta,
|
||
)
|
||
|
||
|
||
def get_screening_history(hours=24, limit=100):
|
||
"""获取最近N小时的筛选记录"""
|
||
conn = get_conn()
|
||
rows = conn.execute("""
|
||
SELECT * FROM screening_log
|
||
WHERE layer='细筛' AND julianday(?) - julianday(scan_time) < ?
|
||
ORDER BY score DESC, scan_time DESC LIMIT ?
|
||
""", (datetime.now().isoformat(), hours / 24.0, limit)).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def get_stats():
|
||
"""兼容导出：统计聚合已迁移到 analytics 模块。"""
|
||
from app.db.analytics import get_stats as _get_stats
|
||
|
||
return _get_stats()
|
||
|
||
|
||
# ==================== 原有状态跟踪（兼容） ====================
|
||
|
||
def update_state(symbol, new_state, score=0, anomaly_type="", sector="",
|
||
leader_status="", detail=None):
|
||
"""更新币状态（兼容旧接口）"""
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT * FROM coin_state WHERE symbol=?", (symbol,)).fetchone()
|
||
|
||
if row:
|
||
old_state = row["state"]
|
||
old_score = row["score"]
|
||
last_alert_time = row["last_alert_time"] or ""
|
||
last_alert_level = row["last_alert_level"] or ""
|
||
|
||
# 状态升级逻辑
|
||
state_order = {"过期": 0, "蓄力": 1, "加速": 2, "爆发": 3}
|
||
should_alert = False
|
||
alert_level = "low"
|
||
|
||
if state_order.get(new_state, 0) > state_order.get(old_state, 0):
|
||
should_alert = True
|
||
alert_level = "high" if new_state == "爆发" else "medium" if new_state == "加速" else "low"
|
||
elif new_state == old_state:
|
||
# 同级别：检查状态冷却，避免频繁重复推荐
|
||
cooldown_hours = confirm_state_cooldown_hours()
|
||
if last_alert_time:
|
||
try:
|
||
last_dt = datetime.fromisoformat(last_alert_time)
|
||
hours_since = (datetime.now() - last_dt).total_seconds() / 3600
|
||
if hours_since < cooldown_hours:
|
||
conn.close()
|
||
return {"should_alert": False, "alert_level": "none", "reason": f"状态冷却中({hours_since:.1f}h<{cooldown_hours}h)"}
|
||
except Exception:
|
||
pass
|
||
if score > old_score:
|
||
# 同级别分数提升，检查冷却
|
||
if not last_alert_time:
|
||
should_alert = True
|
||
alert_level = "medium"
|
||
else:
|
||
try:
|
||
last_dt = datetime.fromisoformat(last_alert_time)
|
||
hours_since = (datetime.now() - last_dt).total_seconds() / 3600
|
||
cooldown = 12 # 12h同币不重复推送同级别
|
||
if hours_since >= cooldown:
|
||
should_alert = True
|
||
alert_level = "medium"
|
||
except Exception:
|
||
should_alert = True
|
||
alert_level = "medium"
|
||
|
||
conn.execute("""
|
||
UPDATE coin_state SET state=?, score=?, anomaly_type=?, sector=?,
|
||
leader_status=?, detected_at=?, detail_json=?
|
||
WHERE symbol=?
|
||
""", (
|
||
new_state, score, anomaly_type, sector, leader_status,
|
||
datetime.now().isoformat(),
|
||
json.dumps(detail, ensure_ascii=False, default=str) if detail else "{}",
|
||
symbol,
|
||
))
|
||
|
||
if should_alert:
|
||
conn.execute("""
|
||
UPDATE coin_state SET last_alert_time=?, last_alert_level=? WHERE symbol=?
|
||
""", (datetime.now().isoformat(), alert_level, symbol))
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
return {"should_alert": should_alert, "alert_level": alert_level}
|
||
else:
|
||
# 新币，首次检测
|
||
conn.execute("""
|
||
INSERT INTO coin_state (symbol, state, score, anomaly_type, sector, leader_status, detected_at, last_alert_time, last_alert_level, detail_json)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
symbol, new_state, score, anomaly_type, sector, leader_status,
|
||
datetime.now().isoformat(), datetime.now().isoformat(), "low",
|
||
json.dumps(detail, ensure_ascii=False, default=str) if detail else "{}",
|
||
))
|
||
conn.commit()
|
||
conn.close()
|
||
return {"should_alert": True, "alert_level": "low"}
|
||
|
||
|
||
def get_candidates_for_confirm():
|
||
"""获取需要确认层检查的候选（加速状态+score≥5）"""
|
||
conn = get_conn()
|
||
rows = conn.execute("""
|
||
SELECT * FROM coin_state WHERE state IN ('加速', '蓄力') AND score >= 5
|
||
""").fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def get_all_active():
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT * FROM coin_state WHERE state != '过期'").fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def expire_old_states(hours=24):
|
||
conn = get_conn()
|
||
conn.execute("""
|
||
UPDATE coin_state SET state='过期' WHERE state != '过期'
|
||
AND julianday(?) - julianday(detected_at) > ?
|
||
""", (datetime.now().isoformat(), hours / 24.0))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
# ==================== 复盘相关 ====================
|
||
|
||
def record_review(rec_id, symbol, outcome, pnl_48h, max_pnl_48h,
|
||
triggered_signals, hit_signals, miss_signals, lesson):
|
||
"""写入一条复盘记录"""
|
||
conn = get_conn()
|
||
conn.execute("""
|
||
INSERT INTO review_log (rec_id, symbol, review_time, outcome, pnl_48h, max_pnl_48h,
|
||
triggered_signals, hit_signals, miss_signals, lesson)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (rec_id, symbol, datetime.now().isoformat(), outcome, pnl_48h, max_pnl_48h,
|
||
json.dumps(triggered_signals, ensure_ascii=False) if isinstance(triggered_signals, list) else triggered_signals,
|
||
json.dumps(hit_signals, ensure_ascii=False) if isinstance(hit_signals, list) else hit_signals,
|
||
json.dumps(miss_signals, ensure_ascii=False) if isinstance(miss_signals, list) else miss_signals,
|
||
lesson))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def update_signal_performance(signal_type, category, is_hit, pnl):
|
||
"""更新信号绩效统计（每次复盘后调用）"""
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT * FROM signal_performance WHERE signal_type=?", (signal_type,)).fetchone()
|
||
|
||
if row:
|
||
total = row["total_count"] + 1
|
||
hits = row["hit_count"] + (1 if is_hit else 0)
|
||
misses = row["miss_count"] + (0 if is_hit else 1)
|
||
old_avg_pnl = row["avg_pnl"]
|
||
# 滚动平均
|
||
new_avg_pnl = round((old_avg_pnl * (total - 1) + pnl) / total, 2)
|
||
hit_rate = round(hits / total * 100, 1) if total > 0 else 0
|
||
|
||
conn.execute("""
|
||
UPDATE signal_performance SET total_count=?, hit_count=?, miss_count=?,
|
||
hit_rate=?, avg_pnl=?, weight=?, last_updated=?
|
||
WHERE signal_type=?
|
||
""", (total, hits, misses, hit_rate, new_avg_pnl, hit_rate / 50, datetime.now().isoformat(), signal_type))
|
||
else:
|
||
conn.execute("""
|
||
INSERT INTO signal_performance (signal_type, category, total_count, hit_count, miss_count,
|
||
hit_rate, avg_pnl, weight, last_updated)
|
||
VALUES (?, ?, 1, ?, ?, ?, ?, ?, ?)
|
||
""", (signal_type, category, 1 if is_hit else 0, 0 if is_hit else 1,
|
||
100 if is_hit else 0, pnl, 2.0 if is_hit else 0, datetime.now().isoformat()))
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_signal_weights():
|
||
"""获取所有信号的当前权重（screener动态调权用）"""
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT signal_type, category, weight, hit_rate, avg_pnl, total_count FROM signal_performance").fetchall()
|
||
conn.close()
|
||
return {row["signal_type"]: dict(row) for row in rows}
|
||
|
||
|
||
def record_missed_explosion(symbol, price_at_detect, price_before, gain_pct,
|
||
reason_missed, features_detected, lesson):
|
||
"""写入一条漏选复盘记录"""
|
||
conn = get_conn()
|
||
conn.execute("""
|
||
INSERT INTO missed_explosions (symbol, detect_time, price_at_detect, price_before,
|
||
gain_pct, reason_missed, features_detected, lesson)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (symbol, datetime.now().isoformat(), price_at_detect, price_before, gain_pct,
|
||
json.dumps(reason_missed, ensure_ascii=False) if isinstance(reason_missed, list) else reason_missed,
|
||
json.dumps(features_detected, ensure_ascii=False) if isinstance(features_detected, list) else features_detected,
|
||
lesson))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_review_stats():
|
||
"""兼容导出：复盘统计已迁移到 analytics 模块。"""
|
||
from app.db.analytics import get_review_stats as _get_review_stats
|
||
|
||
return _get_review_stats(
|
||
conn_provider=get_conn,
|
||
iteration_logs_getter=get_strategy_iteration_logs,
|
||
iteration_summary_getter=get_strategy_iteration_summary,
|
||
)
|
||
|
||
|
||
def _loads_json_field(value, fallback):
|
||
try:
|
||
return json.loads(value) if isinstance(value, str) else (value if value is not None else fallback)
|
||
except Exception:
|
||
return fallback
|
||
|
||
|
||
def log_strategy_iteration(run_date=None, trigger_source="daily_review", title="", summary="",
|
||
findings=None, problems=None, actions=None, changed_rules=None,
|
||
metrics=None, related_symbols=None, config_diff=None, effect_summary=None,
|
||
pollution_summary=None,
|
||
strategy_version="", version_change_summary="",
|
||
success_analysis=None, failure_analysis=None, candidate_rules=None,
|
||
release_decision="", release_reason="", confidence_level="",
|
||
promotion_state="research_only"):
|
||
"""兼容导出：策略迭代写入已迁移到 review_queries 模块。"""
|
||
from app.db.review_queries import log_strategy_iteration as _log_strategy_iteration
|
||
|
||
return _log_strategy_iteration(
|
||
run_date=run_date,
|
||
trigger_source=trigger_source,
|
||
title=title,
|
||
summary=summary,
|
||
findings=findings,
|
||
problems=problems,
|
||
actions=actions,
|
||
changed_rules=changed_rules,
|
||
metrics=metrics,
|
||
related_symbols=related_symbols,
|
||
config_diff=config_diff,
|
||
effect_summary=effect_summary,
|
||
pollution_summary=pollution_summary,
|
||
strategy_version=strategy_version,
|
||
version_change_summary=version_change_summary,
|
||
success_analysis=success_analysis,
|
||
failure_analysis=failure_analysis,
|
||
candidate_rules=candidate_rules,
|
||
release_decision=release_decision,
|
||
release_reason=release_reason,
|
||
confidence_level=confidence_level,
|
||
promotion_state=promotion_state,
|
||
conn_provider=get_conn,
|
||
)
|
||
|
||
|
||
def get_strategy_iteration_logs(limit=30):
|
||
"""兼容导出：策略迭代日志查询已迁移到 review_queries 模块。"""
|
||
from app.db.review_queries import get_strategy_iteration_logs as _get_strategy_iteration_logs
|
||
|
||
return _get_strategy_iteration_logs(limit=limit, conn_provider=get_conn, json_loader=_loads_json_field)
|
||
|
||
|
||
|
||
|
||
def upsert_strategy_rule_candidate(source, rule_type, signal_name, rule_description,
|
||
support_count=0, success_count=0, fail_count=0,
|
||
avg_pnl=0, max_gain=0, max_drawdown=0,
|
||
confidence_score=0, sample_size=0, status="candidate",
|
||
release_version="", notes="", source_ref=""):
|
||
"""新增或更新候选规则。研究结论先沉淀到候选池，避免样本不足时污染主策略。"""
|
||
conn = get_conn()
|
||
now = datetime.now().isoformat()
|
||
existing = conn.execute("""
|
||
SELECT id FROM strategy_rule_candidate
|
||
WHERE source=? AND rule_type=? AND signal_name=? AND rule_description=?
|
||
ORDER BY id DESC LIMIT 1
|
||
""", (source or "", rule_type or "", signal_name or "", rule_description or "")).fetchone()
|
||
if existing:
|
||
conn.execute("""
|
||
UPDATE strategy_rule_candidate
|
||
SET support_count=?, success_count=?, fail_count=?, avg_pnl=?, max_gain=?,
|
||
max_drawdown=?, confidence_score=?, sample_size=?, status=?,
|
||
release_version=?, notes=?, source_ref=COALESCE(NULLIF(?, ''), source_ref), created_at=?
|
||
WHERE id=?
|
||
""", (support_count, success_count, fail_count, avg_pnl, max_gain, max_drawdown,
|
||
confidence_score, sample_size, status, release_version or "", notes or "", source_ref or "", now, existing["id"]))
|
||
candidate_id = existing["id"]
|
||
else:
|
||
cur = conn.execute("""
|
||
INSERT INTO strategy_rule_candidate (
|
||
created_at, source, rule_type, signal_name, rule_description,
|
||
support_count, success_count, fail_count, avg_pnl, max_gain, max_drawdown,
|
||
confidence_score, sample_size, status, release_version, notes, source_ref
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (now, source or "", rule_type or "", signal_name or "", rule_description or "",
|
||
support_count, success_count, fail_count, avg_pnl, max_gain, max_drawdown,
|
||
confidence_score, sample_size, status, release_version or "", notes or "", source_ref or ""))
|
||
candidate_id = cur.lastrowid
|
||
conn.commit()
|
||
conn.close()
|
||
return candidate_id
|
||
|
||
|
||
def record_strategy_failure_pattern(symbol, version="", failure_type="", failure_reason="",
|
||
signal_combo=None, market_context=None,
|
||
entry_quality_issue="", pnl_pct=0, max_drawdown_pct=0, lesson=""):
|
||
"""记录失败模式，用于失败归因统计。"""
|
||
conn = get_conn()
|
||
conn.execute("""
|
||
INSERT INTO strategy_failure_pattern (
|
||
created_at, symbol, version, failure_type, failure_reason, signal_combo,
|
||
market_context_json, entry_quality_issue, pnl_pct, max_drawdown_pct, lesson
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
datetime.now().isoformat(), symbol or "", version or "", failure_type or "",
|
||
failure_reason or "", json.dumps(signal_combo or [], ensure_ascii=False, default=str),
|
||
json.dumps(market_context or {}, ensure_ascii=False, default=str),
|
||
entry_quality_issue or "", pnl_pct or 0, max_drawdown_pct or 0, lesson or "",
|
||
))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_strategy_rule_candidates(limit=50, status=None):
|
||
"""获取候选规则列表。"""
|
||
conn = get_conn()
|
||
params = []
|
||
where = ""
|
||
if status:
|
||
where = "WHERE status=?"
|
||
params.append(status)
|
||
rows = conn.execute(f"""
|
||
SELECT * FROM strategy_rule_candidate
|
||
{where}
|
||
ORDER BY confidence_score DESC, sample_size DESC, created_at DESC
|
||
LIMIT ?
|
||
""", (*params, limit)).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def update_strategy_rule_candidate_status(candidate_id, status, release_version="", notes_append=""):
|
||
"""更新候选规则生命周期状态。"""
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT notes FROM strategy_rule_candidate WHERE id=?", (candidate_id,)).fetchone()
|
||
if not row:
|
||
conn.close()
|
||
return False
|
||
notes = (row["notes"] or "").strip()
|
||
if notes_append:
|
||
notes = (notes + "\n" if notes else "") + f"[{datetime.now().isoformat()}] {notes_append}"
|
||
conn.execute("""
|
||
UPDATE strategy_rule_candidate
|
||
SET status=?, release_version=COALESCE(NULLIF(?, ''), release_version), notes=?, created_at=?
|
||
WHERE id=?
|
||
""", (status or "candidate", release_version or "", notes, datetime.now().isoformat(), candidate_id))
|
||
conn.commit()
|
||
conn.close()
|
||
return True
|
||
|
||
|
||
def get_strategy_failure_patterns(limit=50):
|
||
"""获取失败模式明细。"""
|
||
conn = get_conn()
|
||
rows = conn.execute("""
|
||
SELECT * FROM strategy_failure_pattern
|
||
ORDER BY created_at DESC
|
||
LIMIT ?
|
||
""", (limit,)).fetchall()
|
||
conn.close()
|
||
items = []
|
||
for r in rows:
|
||
item = dict(r)
|
||
item["signal_combo"] = _loads_json_field(item.get("signal_combo"), [])
|
||
item["market_context"] = _loads_json_field(item.get("market_context_json"), {})
|
||
items.append(item)
|
||
return items
|
||
|
||
|
||
def _candidate_signal_key(signal_text):
|
||
"""候选规则归因用的轻量信号归一化。保持与 review_engine._signal_key 语义接近。"""
|
||
text = str(signal_text or "")
|
||
key_map = {
|
||
"量价齐飞": "vp_fly",
|
||
"N倍放量": "vol_Nx",
|
||
"放量": "1h_vol",
|
||
"供需区突破": "zone_break",
|
||
"供给区突破": "zone_break",
|
||
"站稳突破": "zone_break",
|
||
"起爆点": "ignition",
|
||
"静K→动K": "ignition",
|
||
"静K蓄力": "sk_accum",
|
||
"连续3K": "cont3k",
|
||
"连续K": "cont_k",
|
||
"Q≥7": "q7_break",
|
||
"动K": "dyn_k",
|
||
"过期": "stale_signal",
|
||
"历史": "stale_signal",
|
||
"追高": "chase_high",
|
||
"假突破": "false_breakout",
|
||
"量价背离": "vp_divergence",
|
||
}
|
||
for marker, key in key_map.items():
|
||
if marker in text:
|
||
return key
|
||
return text[:12]
|
||
|
||
|
||
|
||
|
||
def _get_factor_recency_fixed_at():
|
||
"""因子时效性修复完成时间：此时间前的推荐视为污染历史参考，不参与正式发布。"""
|
||
try:
|
||
meta = get_meta() or {}
|
||
except Exception:
|
||
meta = {}
|
||
return (meta.get("factor_recency_fixed_at") or meta.get("clean_review_started_at") or "").strip()
|
||
|
||
|
||
def _is_dirty_history_candidate(candidate):
|
||
source = str(candidate.get("source") or "")
|
||
notes = str(candidate.get("notes") or "")
|
||
source_ref = str(candidate.get("source_ref") or "")
|
||
return source in ("history_review_auto", "dirty_history_reference") or "dirty_history" in source_ref or "污染历史" in notes
|
||
|
||
def _candidate_status_for_metrics(rule_type, sample_size, confidence, avg_pnl, current_status="candidate",
|
||
min_gray_samples=10, min_gray_confidence=65):
|
||
"""候选规则生命周期状态判定。"""
|
||
if current_status == "active":
|
||
return "active"
|
||
if sample_size >= min_gray_samples and confidence >= min_gray_confidence and (avg_pnl > 0 or rule_type == "penalty"):
|
||
return "gray"
|
||
if sample_size >= 8 and ((rule_type != "penalty" and confidence < 35) or avg_pnl <= -3):
|
||
return "rejected"
|
||
if current_status in ("gray", "rejected"):
|
||
return current_status
|
||
return "candidate"
|
||
|
||
|
||
|
||
def _classify_failure_type_from_text(review):
|
||
"""历史失败模式回填用的本地分类器，避免 altcoin_db 反向依赖 review_engine。"""
|
||
signals = review.get("triggered_signals") or []
|
||
miss = review.get("miss_signals") or []
|
||
lesson = review.get("lesson") or ""
|
||
text = " ".join([str(x) for x in signals + miss]) + " " + str(lesson or "")
|
||
pnl = float(review.get("pnl_48h") or 0)
|
||
outcome = review.get("outcome") or ""
|
||
if any(k in text for k in ["过期", "历史", "旧放量", "age_bars", "已过期", "小时前", "旧起爆"]):
|
||
return "过期因子误判", "历史放量/起爆/突破不能当作当前触发信号，必须做时效闸门"
|
||
if any(k in text for k in ["假突破", "突破失败", "未站稳", "冲高回落"]):
|
||
return "假突破", "突破后没有站稳或快速回落，需要增加站稳/承接确认"
|
||
if any(k in text for k in ["量价背离", "缩量上涨", "放量下跌", "无量拉升"]):
|
||
return "量价背离", "价格动作与成交量不匹配，量能确认不足"
|
||
if any(k in text for k in ["高位", "追高", "涨幅过大", "乖离"]):
|
||
return "追高风险", "入场位置偏高，盈亏比和回撤风险恶化"
|
||
if any(k in text for k in ["承接不足", "无承接", "上影线", "砸盘"]):
|
||
return "高位无承接", "高位出现抛压但缺少买盘承接"
|
||
if any(k in text for k in ["板块退潮", "热点退潮", "龙头走弱", "板块分歧"]):
|
||
return "板块退潮", "板块热度回落，个币信号容易失效"
|
||
if any(k in text for k in ["BTC", "大盘", "反向共振", "系统性"]):
|
||
return "BTC/大盘反向共振", "大盘方向与个币信号冲突，需要宏观/主流币过滤"
|
||
if any(k in text for k in ["止损", "盈亏比", "RR", "止盈"]):
|
||
return "止损/盈亏比不合理", "止损或止盈结构不合理，导致信号收益风险不匹配"
|
||
if "滞后" in text or "MACD" in text or "RSI" in text:
|
||
return "滞后信号追高", "滞后指标占比高，容易形成事后确认/追高失败"
|
||
if "缺乏前瞻" in text or "前瞻" not in text:
|
||
return "前瞻信号不足", "缺少量价/PA等前瞻性确认"
|
||
if "横盘" in text or outcome == "横盘":
|
||
return "信号强度不足", "触发后未形成有效爆发，确认条件偏弱"
|
||
if "回撤" in text or pnl < -3:
|
||
return "入场点太晚", "入场后回撤/亏损明显，买点可能滞后或确认过慢"
|
||
return "未分类失败", "需要继续积累样本做二级归因"
|
||
|
||
|
||
def backfill_strategy_failure_patterns(limit=2000, dry_run=False):
|
||
"""从历史 review_log 回填失败模式库，按 rec_id 去重。"""
|
||
conn = get_conn()
|
||
rows = conn.execute("""
|
||
SELECT rl.*, r.strategy_version, r.max_drawdown_pct
|
||
FROM review_log rl
|
||
LEFT JOIN recommendation r ON r.id = rl.rec_id
|
||
WHERE rl.outcome IN ('失败','横盘')
|
||
ORDER BY rl.review_time DESC
|
||
LIMIT ?
|
||
""", (limit,)).fetchall()
|
||
existing = set()
|
||
for r in conn.execute("SELECT market_context_json FROM strategy_failure_pattern").fetchall():
|
||
ctx = _loads_json_field(r["market_context_json"], {})
|
||
if ctx.get("rec_id") is not None:
|
||
existing.add(str(ctx.get("rec_id")))
|
||
inserted = 0
|
||
skipped = 0
|
||
type_counts = {}
|
||
examples = []
|
||
for row in rows:
|
||
item = dict(row)
|
||
rec_id = item.get("rec_id")
|
||
if str(rec_id) in existing:
|
||
skipped += 1
|
||
continue
|
||
triggered = _loads_json_field(item.get("triggered_signals"), []) or []
|
||
miss = _loads_json_field(item.get("miss_signals"), []) or []
|
||
item["triggered_signals"] = triggered
|
||
item["miss_signals"] = miss
|
||
ftype, reason = _classify_failure_type_from_text(item)
|
||
type_counts[ftype] = type_counts.get(ftype, 0) + 1
|
||
if len(examples) < 10:
|
||
examples.append({"rec_id": rec_id, "symbol": item.get("symbol"), "failure_type": ftype, "reason": reason})
|
||
if not dry_run:
|
||
conn.execute("""
|
||
INSERT INTO strategy_failure_pattern (
|
||
created_at, symbol, version, failure_type, failure_reason, signal_combo,
|
||
market_context_json, entry_quality_issue, pnl_pct, max_drawdown_pct, lesson
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
datetime.now().isoformat(), item.get("symbol") or "", item.get("strategy_version") or "",
|
||
ftype, reason, json.dumps(triggered, ensure_ascii=False, default=str),
|
||
json.dumps({"source": "history_backfill", "rec_id": rec_id, "outcome": item.get("outcome"), "review_time": item.get("review_time")}, ensure_ascii=False, default=str),
|
||
reason, float(item.get("pnl_48h") or 0), float(item.get("max_drawdown_pct") or 0), item.get("lesson") or "",
|
||
))
|
||
existing.add(str(rec_id))
|
||
inserted += 1
|
||
if not dry_run:
|
||
conn.commit()
|
||
conn.close()
|
||
return {"dry_run": dry_run, "scanned": len(rows), "inserted": inserted, "skipped_existing": skipped, "type_counts": type_counts, "examples": examples}
|
||
|
||
|
||
def generate_candidates_from_review_history(min_samples=20, min_bonus_confidence=55, max_penalty_confidence=40, dry_run=False):
|
||
"""从历史 review_log 自动生成候选规则池。"""
|
||
conn = get_conn()
|
||
rows = conn.execute("""
|
||
SELECT rl.*, r.max_drawdown_pct
|
||
FROM review_log rl
|
||
LEFT JOIN recommendation r ON r.id = rl.rec_id
|
||
ORDER BY rl.review_time DESC
|
||
""").fetchall()
|
||
buckets = {}
|
||
for row in rows:
|
||
item = dict(row)
|
||
triggered = _loads_json_field(item.get("triggered_signals"), []) or []
|
||
hit = _loads_json_field(item.get("hit_signals"), []) or []
|
||
miss = _loads_json_field(item.get("miss_signals"), []) or []
|
||
keys = {_candidate_signal_key(x) for x in list(triggered) + list(hit) + list(miss) if str(x).strip()}
|
||
if not keys:
|
||
continue
|
||
for key in keys:
|
||
b = buckets.setdefault(key, {"sample_size": 0, "success_count": 0, "fail_count": 0, "pnl_values": [], "dd_values": []})
|
||
b["sample_size"] += 1
|
||
if item.get("outcome") == "爆发":
|
||
b["success_count"] += 1
|
||
elif item.get("outcome") in ("失败", "横盘"):
|
||
b["fail_count"] += 1
|
||
b["pnl_values"].append(float(item.get("pnl_48h") or 0))
|
||
b["dd_values"].append(float(item.get("max_drawdown_pct") or 0))
|
||
generated = []
|
||
for key, b in buckets.items():
|
||
sample = b["sample_size"]
|
||
if sample < min_samples:
|
||
continue
|
||
resolved = b["success_count"] + b["fail_count"]
|
||
confidence = round(b["success_count"] / resolved * 100, 1) if resolved else 0
|
||
avg_pnl = round(sum(b["pnl_values"]) / len(b["pnl_values"]), 2) if b["pnl_values"] else 0
|
||
max_gain = round(max(b["pnl_values"]), 2) if b["pnl_values"] else 0
|
||
max_drawdown = round(min(b["dd_values"]), 2) if b["dd_values"] else 0
|
||
rule_type = "bonus" if confidence >= min_bonus_confidence and avg_pnl > 0 else "penalty" if confidence <= max_penalty_confidence else "observe"
|
||
if rule_type == "observe":
|
||
continue
|
||
status = _candidate_status_for_metrics(rule_type, sample, confidence, avg_pnl, "candidate")
|
||
if rule_type == "bonus":
|
||
desc = f"历史样本候选加分因子：{key}，样本{sample}，成功{b['success_count']}，失败/横盘{b['fail_count']}，置信{confidence}%，均值{avg_pnl}%"
|
||
else:
|
||
desc = f"历史样本候选惩罚因子：{key}，样本{sample}，成功{b['success_count']}，失败/横盘{b['fail_count']}，置信{confidence}%，均值{avg_pnl}%"
|
||
candidate = {
|
||
"source": "dirty_history_reference",
|
||
"rule_type": rule_type,
|
||
"signal_name": key,
|
||
"rule_description": desc,
|
||
"support_count": sample,
|
||
"success_count": b["success_count"],
|
||
"fail_count": b["fail_count"],
|
||
"avg_pnl": avg_pnl,
|
||
"max_gain": max_gain,
|
||
"max_drawdown": max_drawdown,
|
||
"confidence_score": confidence,
|
||
"sample_size": sample,
|
||
"status": status,
|
||
"source_ref": f"dirty_history:{key}",
|
||
}
|
||
generated.append(candidate)
|
||
if not dry_run:
|
||
upsert_strategy_rule_candidate(
|
||
source=candidate["source"], rule_type=rule_type, signal_name=key,
|
||
rule_description=desc, support_count=sample, success_count=b["success_count"],
|
||
fail_count=b["fail_count"], avg_pnl=avg_pnl, max_gain=max_gain,
|
||
max_drawdown=max_drawdown, confidence_score=confidence, sample_size=sample,
|
||
status=status, notes="历史review_log自动生成：候选规则仍需灰度验证后才可发布",
|
||
source_ref=candidate["source_ref"],
|
||
)
|
||
if not dry_run:
|
||
conn.commit()
|
||
conn.close()
|
||
generated.sort(key=lambda x: (-x["sample_size"], x["rule_type"], -x["confidence_score"]))
|
||
return {"dry_run": dry_run, "review_rows": len(rows), "generated_count": len(generated), "generated": generated[:80]}
|
||
|
||
def dry_run_strategy_candidate_performance(min_gray_samples=10, min_gray_confidence=65):
|
||
"""不写库的候选规则表现 dry-run，用于复盘系统验收。"""
|
||
conn = get_conn()
|
||
candidates = [dict(r) for r in conn.execute("SELECT * FROM strategy_rule_candidate").fetchall()]
|
||
clean_started_at = _get_factor_recency_fixed_at()
|
||
if clean_started_at:
|
||
review_rows = conn.execute("""
|
||
SELECT rl.*, r.strategy_version, r.max_drawdown_pct, r.rec_time
|
||
FROM review_log rl
|
||
LEFT JOIN recommendation r ON r.id = rl.rec_id
|
||
WHERE r.rec_time >= ?
|
||
ORDER BY rl.review_time DESC
|
||
""", (clean_started_at,)).fetchall()
|
||
else:
|
||
review_rows = conn.execute("""
|
||
SELECT rl.*, r.strategy_version, r.max_drawdown_pct, r.rec_time
|
||
FROM review_log rl
|
||
LEFT JOIN recommendation r ON r.id = rl.rec_id
|
||
ORDER BY rl.review_time DESC
|
||
""").fetchall()
|
||
failure_rows = [dict(r) for r in conn.execute("SELECT * FROM strategy_failure_pattern ORDER BY created_at DESC").fetchall()]
|
||
try:
|
||
current_version = str(get_meta().get("strategy_version") or "").strip()
|
||
except Exception:
|
||
current_version = ""
|
||
conn.close()
|
||
|
||
review_items = []
|
||
for row in review_rows:
|
||
item = dict(row)
|
||
triggered = _loads_json_field(item.get("triggered_signals"), []) or []
|
||
hit = _loads_json_field(item.get("hit_signals"), []) or []
|
||
miss = _loads_json_field(item.get("miss_signals"), []) or []
|
||
all_sigs = list(triggered) + list(hit) + list(miss)
|
||
item["signal_keys"] = {_candidate_signal_key(x) for x in all_sigs}
|
||
item["all_signal_text"] = " ".join(str(x) for x in all_sigs)
|
||
review_items.append(item)
|
||
|
||
evaluated = []
|
||
for c in candidates:
|
||
status = c.get("status") or "candidate"
|
||
source = c.get("source") or ""
|
||
rule_type = c.get("rule_type") or ""
|
||
signal_name = c.get("signal_name") or ""
|
||
source_ref = c.get("source_ref") or ""
|
||
dirty_history = _is_dirty_history_candidate(c)
|
||
if dirty_history:
|
||
evaluated.append({**c, "sample_size": 0, "support_count": 0, "success_count": 0, "fail_count": 0,
|
||
"dry_run_status": "dirty_history", "release_gate_passed": False,
|
||
"gate_reason": "因子时效修复前的污染历史参考：不参与干净样本统计，不允许发布"})
|
||
continue
|
||
if status == "active":
|
||
evaluated.append({**c, "dry_run_status": "active", "release_gate_passed": True, "gate_reason": "已正式生效，不参与dry-run降级"})
|
||
continue
|
||
if source.startswith("dual_attribution_failure") or source_ref.startswith("failure:") or rule_type == "penalty":
|
||
ftype = signal_name or source_ref.replace("failure:", "")
|
||
matched = [r for r in failure_rows if (r.get("failure_type") or "") == ftype or ftype in (r.get("failure_reason") or "")]
|
||
sample_size = len(matched)
|
||
success_count = 0
|
||
fail_count = sample_size
|
||
pnl_values = [float(r.get("pnl_pct") or 0) for r in matched]
|
||
dd_values = [float(r.get("max_drawdown_pct") or 0) for r in matched]
|
||
confidence = round(min(95, 45 + fail_count * 8), 1) if sample_size else float(c.get("confidence_score") or 0)
|
||
else:
|
||
key = signal_name or source_ref.replace("review:", "")
|
||
matched = [item for item in review_items if key and (key in item["signal_keys"] or key in item["all_signal_text"] or signal_name in item["all_signal_text"])]
|
||
sample_size = len(matched)
|
||
success_count = sum(1 for r in matched if r.get("outcome") == "爆发")
|
||
fail_count = sum(1 for r in matched if r.get("outcome") in ("失败", "横盘"))
|
||
pnl_values = [float(r.get("pnl_48h") or 0) for r in matched]
|
||
dd_values = [float(r.get("max_drawdown_pct") or 0) for r in matched]
|
||
resolved = success_count + fail_count
|
||
confidence = round(success_count / resolved * 100, 1) if resolved else float(c.get("confidence_score") or 0)
|
||
avg_pnl = round(sum(pnl_values) / len(pnl_values), 2) if pnl_values else float(c.get("avg_pnl") or 0)
|
||
max_gain = round(max(pnl_values), 2) if pnl_values else float(c.get("max_gain") or 0)
|
||
max_drawdown = round(min(dd_values), 2) if dd_values else float(c.get("max_drawdown") or 0)
|
||
dry_status = _candidate_status_for_metrics(rule_type, sample_size, confidence, avg_pnl, status, min_gray_samples, min_gray_confidence)
|
||
gate_passed = dry_status in ("gray", "active")
|
||
if dry_status == "gray":
|
||
gate_reason = f"样本{sample_size}≥{min_gray_samples}，置信{confidence}%≥{min_gray_confidence}，avg_pnl={avg_pnl}%：可进入灰度，仍不升版"
|
||
elif dry_status == "rejected":
|
||
gate_reason = f"样本{sample_size}已足够但置信/收益不达标：淘汰，不允许发布"
|
||
else:
|
||
gate_reason = f"样本{sample_size}或置信{confidence}%不足：只研究不发布"
|
||
evaluated.append({
|
||
**c,
|
||
"sample_size": sample_size,
|
||
"support_count": sample_size,
|
||
"success_count": success_count,
|
||
"fail_count": fail_count,
|
||
"avg_pnl": avg_pnl,
|
||
"max_gain": max_gain,
|
||
"max_drawdown": max_drawdown,
|
||
"confidence_score": confidence,
|
||
"dry_run_status": dry_status,
|
||
"release_gate_passed": gate_passed,
|
||
"gate_reason": gate_reason,
|
||
})
|
||
|
||
gray_ready = [x for x in evaluated if x.get("dry_run_status") == "gray"]
|
||
active_ready = [x for x in evaluated if x.get("dry_run_status") == "active"]
|
||
rejected = [x for x in evaluated if x.get("dry_run_status") == "rejected"]
|
||
can_release = False
|
||
release_reason = "dry-run只评估候选规则表现，不执行 learned_rules 写入或版本升级"
|
||
return {
|
||
"dry_run": True,
|
||
"current_version": current_version,
|
||
"review_sample_count": len(review_items),
|
||
"clean_started_at": clean_started_at,
|
||
"sample_window": "clean_after_factor_recency_fix" if clean_started_at else "all_history",
|
||
"dirty_history_candidate_count": sum(1 for c in candidates if _is_dirty_history_candidate(c)),
|
||
"candidate_count": len(candidates),
|
||
"gray_ready_count": len(gray_ready),
|
||
"active_count": len(active_ready),
|
||
"rejected_count": len(rejected),
|
||
"would_bump_version": can_release,
|
||
"release_reason": release_reason,
|
||
"gate_policy": {
|
||
"gray": f"sample_size≥{min_gray_samples} 且 confidence≥{min_gray_confidence} 且 avg_pnl>0（penalty规则可不要求avg_pnl>0）",
|
||
"reject": "sample_size≥8 且 confidence<35 或 avg_pnl≤-3",
|
||
"release": "dry-run不发布；正式发布仍由复盘发布闸门统一控制",
|
||
},
|
||
"evaluated_candidates": sorted(evaluated, key=lambda x: (x.get("dry_run_status") != "gray", -float(x.get("sample_size") or 0), -float(x.get("confidence_score") or 0)))[:80],
|
||
}
|
||
|
||
def refresh_strategy_candidate_performance(min_gray_samples=10, min_gray_confidence=65):
|
||
"""刷新候选规则表现。
|
||
|
||
用 review_log + recommendation 复盘结果回填候选规则的 sample/success/fail/avg_pnl，
|
||
并按门槛自动 candidate/gray/rejected，active 不降级。
|
||
"""
|
||
conn = get_conn()
|
||
candidates = conn.execute("SELECT * FROM strategy_rule_candidate").fetchall()
|
||
clean_started_at = _get_factor_recency_fixed_at()
|
||
if clean_started_at:
|
||
review_rows = conn.execute("""
|
||
SELECT rl.*, r.strategy_version, r.max_drawdown_pct, r.rec_time
|
||
FROM review_log rl
|
||
LEFT JOIN recommendation r ON r.id = rl.rec_id
|
||
WHERE r.rec_time >= ?
|
||
ORDER BY rl.review_time DESC
|
||
""", (clean_started_at,)).fetchall()
|
||
else:
|
||
review_rows = conn.execute("""
|
||
SELECT rl.*, r.strategy_version, r.max_drawdown_pct, r.rec_time
|
||
FROM review_log rl
|
||
LEFT JOIN recommendation r ON r.id = rl.rec_id
|
||
ORDER BY rl.review_time DESC
|
||
""").fetchall()
|
||
failure_rows = conn.execute("SELECT * FROM strategy_failure_pattern ORDER BY created_at DESC").fetchall()
|
||
|
||
def loads(value, fallback):
|
||
return _loads_json_field(value, fallback)
|
||
|
||
review_items = []
|
||
for row in review_rows:
|
||
item = dict(row)
|
||
triggered = loads(item.get("triggered_signals"), []) or []
|
||
hit = loads(item.get("hit_signals"), []) or []
|
||
miss = loads(item.get("miss_signals"), []) or []
|
||
all_sigs = list(triggered) + list(hit) + list(miss)
|
||
item["signal_keys"] = {_candidate_signal_key(x) for x in all_sigs}
|
||
item["all_signal_text"] = " ".join(str(x) for x in all_sigs)
|
||
review_items.append(item)
|
||
|
||
updated = []
|
||
for cand in candidates:
|
||
c = dict(cand)
|
||
cid = c["id"]
|
||
status = c.get("status") or "candidate"
|
||
if status == "active":
|
||
continue
|
||
source = c.get("source") or ""
|
||
rule_type = c.get("rule_type") or ""
|
||
signal_name = c.get("signal_name") or ""
|
||
source_ref = c.get("source_ref") or ""
|
||
desc = c.get("rule_description") or ""
|
||
if _is_dirty_history_candidate(c):
|
||
updated.append({
|
||
"id": cid, "signal_name": signal_name, "source": source, "rule_type": rule_type,
|
||
"sample_size": 0, "success_count": 0, "fail_count": 0, "confidence_score": c.get("confidence_score") or 0,
|
||
"avg_pnl": c.get("avg_pnl") or 0, "status": "dirty_history", "description": desc,
|
||
"gate_reason": "污染历史参考，不参与干净样本刷新",
|
||
})
|
||
continue
|
||
|
||
matched = []
|
||
if source.startswith("dual_attribution_failure") or source_ref.startswith("failure:") or rule_type == "penalty":
|
||
ftype = signal_name or source_ref.replace("failure:", "")
|
||
frows = [dict(r) for r in failure_rows if (r["failure_type"] or "") == ftype or ftype in (r["failure_reason"] or "")]
|
||
sample_size = len(frows)
|
||
success_count = 0
|
||
fail_count = sample_size
|
||
pnl_values = [float(r.get("pnl_pct") or 0) for r in frows]
|
||
dd_values = [float(r.get("max_drawdown_pct") or 0) for r in frows]
|
||
confidence = round(min(95, 45 + fail_count * 8), 1) if sample_size else float(c.get("confidence_score") or 0)
|
||
else:
|
||
key = signal_name or source_ref.replace("review:", "")
|
||
for item in review_items:
|
||
if key and (key in item["signal_keys"] or key in item["all_signal_text"] or signal_name in item["all_signal_text"]):
|
||
matched.append(item)
|
||
sample_size = len(matched)
|
||
success_count = sum(1 for r in matched if r.get("outcome") == "爆发")
|
||
fail_count = sum(1 for r in matched if r.get("outcome") in ("失败", "横盘"))
|
||
pnl_values = [float(r.get("pnl_48h") or 0) for r in matched]
|
||
dd_values = [float(r.get("max_drawdown_pct") or 0) for r in matched]
|
||
resolved = success_count + fail_count
|
||
confidence = round(success_count / resolved * 100, 1) if resolved else float(c.get("confidence_score") or 0)
|
||
|
||
avg_pnl = round(sum(pnl_values) / len(pnl_values), 2) if pnl_values else float(c.get("avg_pnl") or 0)
|
||
max_gain = round(max(pnl_values), 2) if pnl_values else float(c.get("max_gain") or 0)
|
||
max_drawdown = round(min(dd_values), 2) if dd_values else float(c.get("max_drawdown") or 0)
|
||
|
||
new_status = _candidate_status_for_metrics(
|
||
rule_type, sample_size, confidence, avg_pnl, status,
|
||
min_gray_samples=min_gray_samples, min_gray_confidence=min_gray_confidence,
|
||
)
|
||
|
||
note = (c.get("notes") or "").strip()
|
||
audit_note = f"[{datetime.now().isoformat()}] 自动评估: 样本{sample_size}, 成功{success_count}, 失败{fail_count}, 置信{confidence}%, avg_pnl={avg_pnl}%, status={new_status}"
|
||
if audit_note not in note:
|
||
note = (note + "\n" if note else "") + audit_note
|
||
|
||
conn.execute("""
|
||
UPDATE strategy_rule_candidate
|
||
SET support_count=?, success_count=?, fail_count=?, avg_pnl=?, max_gain=?,
|
||
max_drawdown=?, confidence_score=?, sample_size=?, status=?, notes=?, created_at=?
|
||
WHERE id=?
|
||
""", (sample_size, success_count, fail_count, avg_pnl, max_gain, max_drawdown,
|
||
confidence, sample_size, new_status, note, datetime.now().isoformat(), cid))
|
||
updated.append({
|
||
"id": cid,
|
||
"signal_name": signal_name,
|
||
"source": source,
|
||
"rule_type": rule_type,
|
||
"sample_size": sample_size,
|
||
"success_count": success_count,
|
||
"fail_count": fail_count,
|
||
"confidence_score": confidence,
|
||
"avg_pnl": avg_pnl,
|
||
"status": new_status,
|
||
"description": desc,
|
||
})
|
||
conn.commit()
|
||
conn.close()
|
||
return updated
|
||
|
||
|
||
def get_strategy_iteration_dashboard(days=30):
|
||
"""迭代页聚合数据：总览 + 候选规则 + 失败模式 + 时间线。"""
|
||
summary = get_strategy_iteration_summary(days=days)
|
||
candidates = get_strategy_rule_candidates(limit=80)
|
||
failures = get_strategy_failure_patterns(limit=80)
|
||
logs = get_strategy_iteration_logs(limit=40)
|
||
status_counts = {}
|
||
source_counts = {}
|
||
for c in candidates:
|
||
status_counts[c.get("status") or "candidate"] = status_counts.get(c.get("status") or "candidate", 0) + 1
|
||
source_counts[c.get("source") or "unknown"] = source_counts.get(c.get("source") or "unknown", 0) + 1
|
||
failure_counts = {}
|
||
for f in failures:
|
||
ft = f.get("failure_type") or "未分类"
|
||
failure_counts[ft] = failure_counts.get(ft, 0) + 1
|
||
release_counts = {}
|
||
for log in logs:
|
||
rd = log.get("release_decision") or "unknown"
|
||
release_counts[rd] = release_counts.get(rd, 0) + 1
|
||
dry_run = dry_run_strategy_candidate_performance()
|
||
latest_log = logs[0] if logs else {}
|
||
return {
|
||
"summary": summary,
|
||
"overview": {
|
||
"total_logs": len(logs),
|
||
"candidate_count": len(candidates),
|
||
"candidate_status_counts": status_counts,
|
||
"candidate_source_counts": source_counts,
|
||
"failure_type_counts": [{"type": k, "count": v} for k, v in sorted(failure_counts.items(), key=lambda x: (-x[1], x[0]))],
|
||
"release_decision_counts": release_counts,
|
||
"latest_release_decision": latest_log.get("release_decision") or "hold",
|
||
"latest_release_reason": latest_log.get("release_reason") or latest_log.get("version_change_summary") or "暂无发布决策说明",
|
||
"dry_run_summary": {
|
||
"review_sample_count": dry_run.get("review_sample_count", 0),
|
||
"clean_started_at": dry_run.get("clean_started_at", ""),
|
||
"sample_window": dry_run.get("sample_window", "all_history"),
|
||
"dirty_history_candidate_count": dry_run.get("dirty_history_candidate_count", 0),
|
||
"candidate_count": dry_run.get("candidate_count", 0),
|
||
"gray_ready_count": dry_run.get("gray_ready_count", 0),
|
||
"rejected_count": dry_run.get("rejected_count", 0),
|
||
"would_bump_version": dry_run.get("would_bump_version", False),
|
||
"release_reason": dry_run.get("release_reason", ""),
|
||
},
|
||
},
|
||
"dry_run": dry_run,
|
||
"candidates": candidates,
|
||
"failures": failures,
|
||
"logs": logs,
|
||
}
|
||
|
||
def get_strategy_iteration_summary(days=30):
|
||
"""兼容导出：策略迭代汇总已迁移到 review_queries 模块。"""
|
||
from app.db.review_queries import get_strategy_iteration_summary as _get_strategy_iteration_summary
|
||
|
||
return _get_strategy_iteration_summary(days=days, conn_provider=get_conn, json_loader=_loads_json_field)
|
||
|
||
|
||
def log_cron_run(job_name, script_name, run_status, result_status="", started_at="", finished_at="",
|
||
duration_ms=0, summary=None, error_message=""):
|
||
"""记录一次 cron 运行汇总"""
|
||
conn = get_conn()
|
||
conn.execute("""
|
||
INSERT INTO cron_run_log (
|
||
job_name, script_name, run_status, result_status,
|
||
started_at, finished_at, duration_ms, summary_json, error_message
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
job_name,
|
||
script_name,
|
||
run_status,
|
||
result_status,
|
||
started_at or datetime.now().isoformat(),
|
||
finished_at or datetime.now().isoformat(),
|
||
int(duration_ms or 0),
|
||
json.dumps(summary or {}, ensure_ascii=False, default=str),
|
||
(error_message or "")[:1000],
|
||
))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_cron_run_logs(limit=50, job_name=None):
|
||
"""获取 cron 运行日志列表"""
|
||
conn = get_conn()
|
||
sql = """
|
||
SELECT * FROM cron_run_log
|
||
{where_clause}
|
||
ORDER BY started_at DESC, id DESC
|
||
LIMIT ?
|
||
"""
|
||
params = []
|
||
where_clause = ""
|
||
if job_name:
|
||
where_clause = "WHERE job_name = ?"
|
||
params.append(job_name)
|
||
params.append(limit)
|
||
rows = conn.execute(sql.format(where_clause=where_clause), tuple(params)).fetchall()
|
||
conn.close()
|
||
|
||
result = []
|
||
for row in rows:
|
||
item = dict(row)
|
||
try:
|
||
item["summary_json"] = json.loads(item.get("summary_json") or "{}")
|
||
except Exception:
|
||
item["summary_json"] = {}
|
||
result.append(item)
|
||
return result
|
||
|
||
|
||
def get_cron_run_summary(hours=24):
|
||
"""获取 cron 运行汇总统计"""
|
||
conn = get_conn()
|
||
now_iso = datetime.now().isoformat()
|
||
rows = conn.execute("""
|
||
SELECT * FROM cron_run_log
|
||
WHERE julianday(?) - julianday(started_at) <= ?
|
||
ORDER BY started_at DESC, id DESC
|
||
""", (now_iso, hours / 24.0)).fetchall()
|
||
conn.close()
|
||
|
||
logs = []
|
||
job_stats = {}
|
||
total_runs = 0
|
||
success_runs = 0
|
||
error_runs = 0
|
||
total_duration = 0
|
||
|
||
for row in rows:
|
||
item = dict(row)
|
||
try:
|
||
item["summary_json"] = json.loads(item.get("summary_json") or "{}")
|
||
except Exception:
|
||
item["summary_json"] = {}
|
||
logs.append(item)
|
||
|
||
total_runs += 1
|
||
total_duration += item.get("duration_ms") or 0
|
||
if item.get("run_status") == "success":
|
||
success_runs += 1
|
||
else:
|
||
error_runs += 1
|
||
|
||
job = item.get("job_name") or "unknown"
|
||
stat = job_stats.setdefault(job, {
|
||
"job_name": job,
|
||
"runs": 0,
|
||
"success_runs": 0,
|
||
"error_runs": 0,
|
||
"avg_duration_ms": 0,
|
||
"last_status": "",
|
||
"last_result_status": "",
|
||
"last_started_at": "",
|
||
"last_finished_at": "",
|
||
"last_error_message": "",
|
||
})
|
||
stat["runs"] += 1
|
||
if item.get("run_status") == "success":
|
||
stat["success_runs"] += 1
|
||
else:
|
||
stat["error_runs"] += 1
|
||
stat["avg_duration_ms"] += item.get("duration_ms") or 0
|
||
if not stat["last_started_at"]:
|
||
stat["last_status"] = item.get("run_status", "")
|
||
stat["last_result_status"] = item.get("result_status", "")
|
||
stat["last_started_at"] = item.get("started_at", "")
|
||
stat["last_finished_at"] = item.get("finished_at", "")
|
||
stat["last_error_message"] = item.get("error_message", "")
|
||
|
||
for stat in job_stats.values():
|
||
stat["success_rate"] = round(stat["success_runs"] / stat["runs"] * 100, 1) if stat["runs"] else 0
|
||
stat["avg_duration_ms"] = round(stat["avg_duration_ms"] / stat["runs"]) if stat["runs"] else 0
|
||
|
||
overall = {
|
||
"hours": hours,
|
||
"total_runs": total_runs,
|
||
"success_runs": success_runs,
|
||
"error_runs": error_runs,
|
||
"success_rate": round(success_runs / total_runs * 100, 1) if total_runs else 0,
|
||
"avg_duration_ms": round(total_duration / total_runs) if total_runs else 0,
|
||
}
|
||
|
||
return {
|
||
"overall": overall,
|
||
"job_stats": sorted(job_stats.values(), key=lambda x: x["job_name"]),
|
||
"recent_logs": logs[:20],
|
||
}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
init_db()
|
||
stats = get_stats()
|
||
print(f"DB初始化完成: {stats}")
|
||
|
||
|
||
def _safe_list_json(value):
|
||
try:
|
||
if isinstance(value, list):
|
||
return value
|
||
if isinstance(value, str) and value.strip():
|
||
parsed = json.loads(value)
|
||
return parsed if isinstance(parsed, list) else []
|
||
except Exception:
|
||
pass
|
||
return []
|
||
|
||
|
||
def _safe_dict_json(value):
|
||
try:
|
||
if isinstance(value, dict):
|
||
return value
|
||
if isinstance(value, str) and value.strip():
|
||
parsed = json.loads(value)
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
except Exception:
|
||
pass
|
||
return {}
|
||
|
||
|
||
def get_strategy_insights():
|
||
"""阶段3：策略可信度看板数据 — 总体表现、因子归因、市场环境归因。
|
||
|
||
只统计已出结果的样本，避免策略页把仍在实时看板里的 active 浮亏/回撤
|
||
与历史推荐页的已完成样本混在一起，造成口径不一致。
|
||
"""
|
||
conn = get_conn()
|
||
rows = conn.execute("SELECT * FROM recommendation ORDER BY rec_time DESC").fetchall()
|
||
conn.close()
|
||
raw_items = [dict(r) for r in rows]
|
||
|
||
def outcome(item):
|
||
status = item.get("status") or ""
|
||
if status in ("hit_tp1", "hit_tp2"):
|
||
return "success"
|
||
if status == "stopped_out":
|
||
return "failed"
|
||
if (item.get("max_pnl_pct") or 0) >= 5:
|
||
return "success"
|
||
if (item.get("pnl_pct") or 0) <= -3 or (item.get("max_drawdown_pct") or 0) <= -5:
|
||
return "failed"
|
||
return "pending"
|
||
|
||
items = [x for x in raw_items if outcome(x) in ("success", "failed")]
|
||
|
||
total = len(items)
|
||
success = sum(1 for x in items if outcome(x) == "success")
|
||
failed = sum(1 for x in items if outcome(x) == "failed")
|
||
resolved = success + failed
|
||
pnl_values = [float(x.get("pnl_pct") or 0) for x in items]
|
||
gains = [p for p in pnl_values if p > 0]
|
||
losses = [p for p in pnl_values if p < 0]
|
||
overview = {
|
||
"total_signals": total,
|
||
"resolved_count": resolved,
|
||
"success_count": success,
|
||
"failed_count": failed,
|
||
"pending_count": total - resolved,
|
||
"win_rate_pct": round(success / resolved * 100, 1) if resolved else 0,
|
||
"avg_pnl_pct": round(sum(pnl_values) / len(pnl_values), 2) if pnl_values else 0,
|
||
"avg_gain_pct": round(sum(gains) / len(gains), 2) if gains else 0,
|
||
"avg_loss_pct": round(sum(losses) / len(losses), 2) if losses else 0,
|
||
"max_gain_pct": round(max([float(x.get("max_pnl_pct") or x.get("pnl_pct") or 0) for x in items] or [0]), 2),
|
||
"max_drawdown_pct": round(min([float(x.get("max_drawdown_pct") or 0) for x in items] or [0]), 2),
|
||
}
|
||
|
||
def add_bucket(bucket_map, key, item):
|
||
if not key:
|
||
return
|
||
b = bucket_map.setdefault(key, {"total_count": 0, "success_count": 0, "failed_count": 0, "pending_count": 0, "pnl_values": [], "max_gains": [], "drawdowns": []})
|
||
b["total_count"] += 1
|
||
oc = outcome(item)
|
||
if oc == "success": b["success_count"] += 1
|
||
elif oc == "failed": b["failed_count"] += 1
|
||
else: b["pending_count"] += 1
|
||
b["pnl_values"].append(float(item.get("pnl_pct") or 0))
|
||
b["max_gains"].append(float(item.get("max_pnl_pct") or item.get("pnl_pct") or 0))
|
||
b["drawdowns"].append(float(item.get("max_drawdown_pct") or 0))
|
||
|
||
def env_buckets_from_market_context(mc):
|
||
"""把当前实际存在的 market_context_json 数值字段转成可归因桶。
|
||
|
||
旧版本只读取 btc_trend/market_regime 等枚举字段，但当前入库字段主要是
|
||
change_24h、turnover_acceleration、volume_24h/funding_rate，导致市场环境归因为空。
|
||
"""
|
||
buckets = []
|
||
try:
|
||
change_24h = float(mc.get("change_24h", 0) or 0)
|
||
turn_1h = float(mc.get("turnover_acceleration_1h", 0) or 0)
|
||
turn_4h = float(mc.get("turnover_acceleration_4h", 0) or 0)
|
||
volume_24h = float(mc.get("volume_24h") or mc.get("quote_volume_24h") or 0)
|
||
funding = float(mc.get("funding_rate", 0) or 0)
|
||
except Exception:
|
||
change_24h = turn_1h = turn_4h = volume_24h = funding = 0
|
||
|
||
if change_24h >= 8:
|
||
buckets.append("24h涨幅:强势拉升≥8%")
|
||
elif change_24h >= 3:
|
||
buckets.append("24h涨幅:温和上涨3-8%")
|
||
elif change_24h <= -3:
|
||
buckets.append("24h涨幅:回撤≤-3%")
|
||
else:
|
||
buckets.append("24h涨幅:震荡-3~3%")
|
||
|
||
if turn_1h >= 3:
|
||
buckets.append("1h成交加速:爆量≥3x")
|
||
elif turn_1h >= 1.5:
|
||
buckets.append("1h成交加速:放量1.5-3x")
|
||
elif turn_1h > 0:
|
||
buckets.append("1h成交加速:平量<1.5x")
|
||
|
||
if turn_4h >= 3:
|
||
buckets.append("4h成交加速:爆量≥3x")
|
||
elif turn_4h >= 1.5:
|
||
buckets.append("4h成交加速:放量1.5-3x")
|
||
elif turn_4h > 0:
|
||
buckets.append("4h成交加速:平量<1.5x")
|
||
|
||
if volume_24h >= 100_000_000:
|
||
buckets.append("24h成交额:高流动性≥1亿")
|
||
elif volume_24h >= 10_000_000:
|
||
buckets.append("24h成交额:中等流动性1千万-1亿")
|
||
elif volume_24h > 0:
|
||
buckets.append("24h成交额:低流动性<1千万")
|
||
|
||
if funding >= 0.0005:
|
||
buckets.append("资金费率:多头拥挤")
|
||
elif funding <= -0.0005:
|
||
buckets.append("资金费率:空头拥挤")
|
||
return buckets
|
||
|
||
factor_map = {}
|
||
env_map = {}
|
||
version_map = {}
|
||
for item in items:
|
||
for factor in _safe_list_json(item.get("signals")):
|
||
add_bucket(factor_map, str(factor).strip(), item)
|
||
mc = _safe_dict_json(item.get("market_context_json"))
|
||
added_env = False
|
||
for key in ("btc_trend", "market_regime", "altcoin_regime", "sentiment"):
|
||
if mc.get(key):
|
||
add_bucket(env_map, f"{key}:{mc.get(key)}", item)
|
||
added_env = True
|
||
for bucket in env_buckets_from_market_context(mc):
|
||
add_bucket(env_map, bucket, item)
|
||
added_env = True
|
||
if item.get("strategy_version"):
|
||
add_bucket(version_map, str(item.get("strategy_version")).strip(), item)
|
||
|
||
def version_sort_key(version: str):
|
||
text = str(version or '').strip()
|
||
if text.startswith('v') or text.startswith('V'):
|
||
text = text[1:]
|
||
parts = []
|
||
for chunk in text.replace('-', '.').split('.'):
|
||
if chunk.isdigit():
|
||
parts.append(int(chunk))
|
||
else:
|
||
m = re.match(r'^(\d+)', chunk)
|
||
if m:
|
||
parts.append(int(m.group(1)))
|
||
else:
|
||
parts.append(chunk)
|
||
return tuple(parts)
|
||
|
||
def serialize(name_key, bucket_map, sort_by_version=False):
|
||
rows = []
|
||
for key, b in bucket_map.items():
|
||
resolved_count = b["success_count"] + b["failed_count"]
|
||
rows.append({
|
||
name_key: key,
|
||
"total_count": b["total_count"],
|
||
"success_count": b["success_count"],
|
||
"failed_count": b["failed_count"],
|
||
"pending_count": b["pending_count"],
|
||
"win_rate_pct": round(b["success_count"] / resolved_count * 100, 1) if resolved_count else 0,
|
||
"avg_pnl_pct": round(sum(b["pnl_values"]) / len(b["pnl_values"]), 2) if b["pnl_values"] else 0,
|
||
"max_gain_pct": round(max(b["max_gains"] or [0]), 2),
|
||
"max_drawdown_pct": round(min(b["drawdowns"] or [0]), 2),
|
||
})
|
||
if sort_by_version:
|
||
rows.sort(key=lambda x: (version_sort_key(x[name_key]), x["total_count"], x["win_rate_pct"]), reverse=True)
|
||
else:
|
||
rows.sort(key=lambda x: (-x["total_count"], -x["win_rate_pct"], x[name_key]))
|
||
return rows
|
||
|
||
return {
|
||
"overview": overview,
|
||
"factor_attribution": serialize("factor", factor_map)[:30],
|
||
"market_environment": serialize("environment", env_map)[:20],
|
||
"version_performance": serialize("strategy_version", version_map, sort_by_version=True)[:20],
|
||
}
|