aaron/alphax
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
aaron 2026-05-23 10:17:26 +08:00
parent ea342161fc
commit ecb4076dfe
8 changed files with 530 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
AGENTS.md
View File

@ -100,7 +100,16 @@ AlphaX 是一个以 `Python + FastAPI + PostgreSQL + Docker + 静态 HTML` 组
9. `app/services/review_engine.py`
负责复盘与策略自迭代，包括信号绩效、漏选复盘、规则候选、版本演进。
### 4.1.1 链上数据源
### 4.1.1 因子评分与复盘进化
- `app/core/factor_scoring.py` 是确认层因子评分中心。新增确认加减分不要继续散落写死 `score += N`，应优先通过 `FactorScorer.delta(factor_code, base_delta, evidence=...)` 计算。
- 稳定因子代码来自 `app/core/signal_taxonomy.py`，例如 `vp_fly_1h_current`、`volume_consecutive_1h`、`ignition_d1_current`、`sector_rotation`、`sentiment_resonance`、`top_trader_long`、`risk_reward_bad`。
- `signal_performance` 是复盘后动态权重来源；`review_engine.py` 更新信号绩效后，`config_loader.get_signal_weights()` 会让下一轮筛选/确认读取生效权重。
- 当前确认层已把核心技术因子、资金面因子、板块因子、舆情因子和买点风险因子接入 `FactorScorer`，并在 `market_context.factor_score_breakdown` / `entry_plan.factor_score_breakdown` 中保留因子明细。
- NodeReal 链上因子通过 `app/db/onchain_db.py#get_onchain_factor_context()` 进入确认层，正向事件如 `whale_accumulation`、`smart_money_buying`、`exchange_outflow` 会加分，风险事件如 `exchange_inflow_risk`、`liquidity_remove_risk`、`holder_concentration_risk` 会扣分；这些因子同样受 `signal_performance.weight` 复盘权重约束。
- 后续新增链上、资金、事件、舆情等非 K 线因子时，必须给出稳定 `factor_code`、默认基准权重、证据字段和复盘归因口径，避免只做展示标签而不参与策略进化。
### 4.1.2 链上数据源
- 当前链上主数据源是 NodeReal，入口在 `app/services/nodereal_client.py` 和 `app/services/onchain_monitor.py`。
- 默认只跑 `ALPHAX_ONCHAIN_PROVIDER=nodereal`，并通过 `ALPHAX_NODEREAL_API_KEY` 访问 EVM JSON-RPC / Enhanced API。
@ -410,6 +419,8 @@ docker compose run --rm alphax-web python scripts/postgres/validate_import.py --
- `price_tracker.py` 会为 active 观察池样本更新观察价/PnL，但未触发入场的 watch_pool/wait_pullback 不能触发止盈止损、不能进入 paper trading 收益账本。
- `rec_state` 是发现层状态（如“爆发/加速”），`execution_status`/`trade_stage` 才是交易执行阶段（如 `buy_now`/`wait_pullback`/`observe`），不要把“发现爆发”直接解读成“现在可买”。
- 静K蓄力旁路已要求配置化共振（见 `rules.yaml` 的 `screener.static_accumulation_bypass.require_resonance`），避免单一静K样本淹没确认层；无追高风险的强势榜异动仍可作为发现入口。
- 确认评分不再应被理解为固定技术分；确认层通过 `FactorScorer` 读取复盘后的 `signal_performance.weight`，高胜率因子会升权，低胜率/负收益因子会降权或淘汰。
- 评分因子必须保留 `factor_score_breakdown`，否则复盘无法知道一次推荐具体由哪些因子贡献、哪些因子拖累。
- `paper_trader.py` 只应处理可执行推荐，不能把观察池样本当成已成交。
- `review_engine.py` 的可信度依赖跟踪数据质量；如果 PnL 没更新，复盘结论也会失真。
- `missed_explosions` 历史数据可能存在同一 symbol 多次记录，读模型/KPI 需要保持去重口径，写入侧后续仍建议加唯一性或冷却约束。

29
app/config/config_loader.py
View File

@ -23,11 +23,30 @@ _yaml_cache_mtime = None
# 兼容旧代码中的信号名写法
_SIGNAL_NAME_ALIASES = {
"N倍放量(≥10x)": "N倍放量",
"连续3x放量(≥3根)": "连续3x放量",
"静K→动K转折": "静K动K转折",
"Q≥7供给区突破": "Q7供给区突破",
"1H放量(量价背离)": "1H放量",
"N倍放量(≥10x)": "vp_fly_1h_current",
"N倍放量": "vp_fly_1h_current",
"量价齐飞": "vp_fly_1h_current",
"连续3x放量(≥3根)": "volume_consecutive_1h",
"连续3x放量": "volume_consecutive_1h",
"静K→动K转折": "ignition_1h_current",
"静K动K转折": "ignition_1h_current",
"Q≥7供给区突破": "breakout_pullback_d1",
"Q7供给区突破": "breakout_pullback_d1",
"1H放量(量价背离)": "volume_divergence_1h",
"量价背离": "volume_divergence_1h",
"静K蓄力": "static_accum_4h",
"大户偏多": "top_trader_long",
"舆情共振": "sentiment_resonance",
"板块联动": "sector_rotation",
"DEX 放量": "dex_volume_spike",
"链上成交放量": "dex_volume_spike",
"流动性增加": "liquidity_add",
"交易所流出": "exchange_outflow",
"鲸鱼增持": "whale_accumulation",
"聪明钱买入": "smart_money_buying",
"流动性撤出风险": "liquidity_remove_risk",
"交易所流入风险": "exchange_inflow_risk",
"持仓集中风险": "holder_concentration_risk",
}

165
app/core/factor_scoring.py Normal file
View File

@ -0,0 +1,165 @@
"""Review-aware factor scoring.
This module is the bridge between daily review results and the next screening
run. Strategy code should score named factors here instead of hard-coding every
``score += N`` forever.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from app.config.config_loader import get_signal_weights
from app.core.signal_taxonomy import signal_label_for_code
DEFAULT_FACTOR_WEIGHTS = {
"vp_fly_1h_current": 5.0,
"volume_consecutive_1h": 4.0,
"cex_top_gainer_24h": 2.0,
"volume_divergence_1h": 1.0,
"static_accum_4h": 5.0,
"higher_lows_4h": 2.0,
"compression_surge_4h": 2.0,
"ignition_1h_current": 4.0,
"ignition_4h_current": 3.0,
"ignition_d1_current": 6.0,
"dynamic_k_1h_bull": 3.0,
"dynamic_k_d1_bull": 5.0,
"breakout_pullback_d1": 8.0,
"breakout_pullback_w1": 8.0,
"breakout_15m_current": 3.0,
"pullback_15m_confirm": 2.0,
"strong_resonance_bypass": 3.0,
"entry_quality_gate": 2.0,
"top_trader_long": 1.0,
"sector_rotation": 2.0,
"sentiment_resonance": 2.0,
"dex_volume_spike": 2.0,
"liquidity_add": 1.5,
"exchange_outflow": 2.0,
"whale_accumulation": 2.5,
"smart_money_buying": 2.5,
"liquidity_remove_risk": 2.5,
"exchange_inflow_risk": 2.5,
"holder_concentration_risk": 2.0,
"funding_extreme": 2.0,
"trend_exhaustion": 3.0,
"false_breakout": 5.0,
"high_position_reject": 5.0,
"risk_reward_bad": 2.0,
}
WEIGHT_ALIASES = {
"vp_fly_1h_current": ("量价齐飞", "1H当前量价齐飞"),
"volume_consecutive_1h": ("连续3x放量", "连续3x放量(≥3根)", "1H连续放量"),
"volume_divergence_1h": ("量价背离", "1H量价背离"),
"static_accum_4h": ("静K蓄力", "4H静K蓄力"),
"ignition_1h_current": ("静K动K转折", "静K→动K转折", "1H当前起爆点"),
"ignition_4h_current": ("静K动K转折", "静K→动K转折", "4H当前起爆点"),
"ignition_d1_current": ("静K动K转折", "静K→动K转折", "日线当前起爆点"),
"breakout_15m_current": ("15min当前突破",),
"pullback_15m_confirm": ("15min回踩确认",),
"top_trader_long": ("大户偏多",),
"sector_rotation": ("板块联动",),
"sentiment_resonance": ("舆情共振",),
"dex_volume_spike": ("DEX 放量", "链上成交放量"),
"liquidity_add": ("流动性增加",),
"exchange_outflow": ("交易所流出",),
"whale_accumulation": ("鲸鱼增持",),
"smart_money_buying": ("聪明钱买入",),
"liquidity_remove_risk": ("流动性撤出风险",),
"exchange_inflow_risk": ("交易所流入风险",),
"holder_concentration_risk": ("持仓集中风险",),
"funding_extreme": ("资金费率极端",),
"trend_exhaustion": ("趋势衰减",),
"false_breakout": ("假突破",),
"high_position_reject": ("高位拒绝",),
"risk_reward_bad": ("盈亏比不合格",),
}
def _safe_float(value: Any, default: float = 0.0) -> float:
try:
if value is None or value == "":
return default
return float(value)
except Exception:
return default
def _clamp(value: float, low: float, high: float) -> float:
return max(low, min(high, value))
@dataclass
class FactorScorer:
"""Apply review-adjusted factor weights and keep an audit trail."""
weights: dict[str, Any] = field(default_factory=dict)
breakdown: list[dict[str, Any]] = field(default_factory=list)
@classmethod
def from_runtime(cls) -> "FactorScorer":
try:
weights = get_signal_weights() or {}
except Exception:
weights = {}
return cls(weights=weights)
def _runtime_weight(self, code: str) -> float | None:
keys = [code, signal_label_for_code(code), *WEIGHT_ALIASES.get(code, ())]
for key in keys:
if key in self.weights:
value = self.weights[key]
if isinstance(value, dict):
value = value.get("weight")
return max(0.0, _safe_float(value))
return None
def delta(self, code: str, base: float, *, evidence: str = "", value: Any = None) -> float:
"""Return the review-aware score delta for one factor.
``base`` remains the strategy designer's default score. Once reviews
have enough samples, ``signal_performance.weight`` becomes a multiplier
against the factor's canonical baseline. A reviewed weight of 0 means
the factor is effectively淘汰 for scoring.
"""
base = _safe_float(base)
runtime_weight = self._runtime_weight(code)
baseline = max(0.1, _safe_float(DEFAULT_FACTOR_WEIGHTS.get(code), abs(base) or 1.0))
if runtime_weight is None:
adjusted = base
multiplier = 1.0
source = "base"
else:
multiplier = _clamp(runtime_weight / baseline, 0.0, 1.8)
adjusted = base * multiplier
source = "review_weight"
adjusted = round(adjusted, 3)
self.breakdown.append(
{
"factor_code": code,
"factor_name": signal_label_for_code(code),
"base_delta": base,
"score_delta": adjusted,
"runtime_weight": runtime_weight,
"baseline_weight": baseline,
"multiplier": round(multiplier, 4),
"source": source,
"evidence": evidence,
"value": value,
}
)
return adjusted
def add_existing(self, code: str, observed_score: float, *, evidence: str = "", value: Any = None, cap: float | None = None) -> float:
base = _safe_float(observed_score)
if cap is not None:
base = min(base, _safe_float(cap, base))
return self.delta(code, base, evidence=evidence, value=value)
def summary(self) -> dict[str, Any]:
total = round(sum(_safe_float(item.get("score_delta")) for item in self.breakdown), 3)
return {"total_delta": total, "items": self.breakdown}

2
app/core/signal_taxonomy.py
View File

@ -12,6 +12,7 @@ from typing import Any, Iterable
SIGNAL_CODE_LABELS = {
"vp_fly_1h_current": "1H当前量价齐飞",
"volume_consecutive_1h": "1H连续放量",
"cex_top_gainer_24h": "CEX 24h强势榜异动",
"vp_fly_1h_stale": "1H历史量价齐飞",
"volume_divergence_1h": "1H量价背离",
@ -25,6 +26,7 @@ SIGNAL_CODE_LABELS = {
"dynamic_k_1h_bull": "1H多头动K",
"dynamic_k_d1_bull": "日线多头动K",
"breakout_pullback_d1": "日线突破回踩",
"breakout_pullback_w1": "周线突破回踩",
"breakout_15m_current": "15min当前突破",
"pullback_15m_confirm": "15min回踩确认",
"strong_resonance_bypass": "强共振旁路",

57
app/db/onchain_db.py
View File

@ -799,6 +799,63 @@ def get_onchain_token_detail(symbol, hours=72):
}
def get_onchain_factor_context(symbol, hours=24):
"""Return compact on-chain factor evidence for strategy scoring.
This read model intentionally does not create recommendations. It only
exposes mapped NodeReal facts to the technical confirmation layer.
"""
init_onchain_tables()
symbol = normalize_symbol(symbol)
if not symbol:
return {"symbol": "", "positive_events": [], "risk_events": [], "metrics": {}, "has_data": False}
cutoff = (datetime.now() - timedelta(hours=int(hours or 24))).isoformat()
conn = get_conn()
try:
metric = conn.execute(
"""
SELECT *
FROM onchain_token_metrics
WHERE symbol=%s AND metric_time >= %s
AND COALESCE(source, '') NOT IN ('dexscreener', 'etherscan', 'helius')
ORDER BY metric_time DESC, id DESC
LIMIT 1
""",
(symbol, cutoff),
).fetchone()
rows = conn.execute(
"""
SELECT *
FROM onchain_events
WHERE symbol=%s AND detected_at >= %s
AND COALESCE(source, '') NOT IN ('dexscreener', 'etherscan', 'helius')
ORDER BY detected_at DESC, confidence DESC, value_usd DESC, id DESC
LIMIT 20
""",
(symbol, cutoff),
).fetchall()
finally:
conn.close()
events = [dict(row) for row in rows]
positive = [e for e in events if e.get("direction") == "positive"]
risks = [e for e in events if e.get("direction") == "risk"]
metric_item = dict(metric) if metric else {}
return {
"symbol": symbol,
"hours": int(hours or 24),
"has_data": bool(metric_item or events),
"metrics": metric_item,
"positive_events": positive,
"risk_events": risks,
"event_count": len(events),
"positive_event_count": len(positive),
"risk_event_count": len(risks),
"top_positive": positive[0] if positive else None,
"top_risk": risks[0] if risks else None,
}
def list_onchain_events(limit=50, offset=0, chain="", signal="", status="", hours=24):
init_onchain_tables()
limit = max(1, min(int(limit or 50), 200))

190
app/services/altcoin_confirm.py
View File

@ -48,6 +48,8 @@ from app.core.opportunity_level import (
select_level_stop_loss,
)
from app.core.opportunity_funnel import build_screening_detail
from app.core.factor_scoring import FactorScorer
from app.db.onchain_db import get_onchain_factor_context
from app.config.config_loader import _get_section as _get_cfg_section
from app.core.pa_engine import (
classify_candles, calc_atr, find_supply_demand_zones,
@ -396,6 +398,87 @@ def compute_sector_context(symbol, cand_detail=None):
return ctx
def _onchain_base_delta(event):
code = str((event or {}).get("signal_code") or "")
value_usd = float((event or {}).get("value_usd") or 0)
confidence = float((event or {}).get("confidence") or 0)
base = {
"whale_accumulation": 2.5,
"smart_money_buying": 2.5,
"exchange_outflow": 2.0,
"dex_volume_spike": 2.0,
"liquidity_add": 1.5,
"liquidity_remove_risk": 2.5,
"exchange_inflow_risk": 2.5,
"holder_concentration_risk": 2.0,
}.get(code, 1.0)
if value_usd >= 1_000_000 or confidence >= 85:
base += 0.5
return min(base, 3.5)
def _apply_onchain_factor_score(symbol, factor_scorer):
"""Score mapped NodeReal evidence as a first-class strategy factor."""
try:
ctx = get_onchain_factor_context(symbol, hours=24)
except Exception:
return 0.0, [], {"has_data": False, "error": "onchain_context_unavailable"}
if not ctx.get("has_data"):
return 0.0, [], ctx
score_delta = 0.0
signals = []
metric = ctx.get("metrics") or {}
onchain_score = float(metric.get("onchain_score") or 0)
risk_score = float(metric.get("risk_score") or 0)
positive_events = ctx.get("positive_events") or []
risk_events = ctx.get("risk_events") or []
for event in positive_events[:3]:
code = event.get("signal_code") or "unknown"
label = event.get("signal_label") or code
delta = factor_scorer.delta(
code,
_onchain_base_delta(event),
evidence=f"NodeReal正向链上事件: {label}",
value={"value_usd": event.get("value_usd"), "confidence": event.get("confidence")},
)
score_delta += delta
signals.append(f"链上正向: {label}(置信{event.get('confidence') or 0}, ${float(event.get('value_usd') or 0):.0f})")
for event in risk_events[:3]:
code = event.get("signal_code") or "holder_concentration_risk"
label = event.get("signal_label") or code
delta = factor_scorer.delta(
code,
_onchain_base_delta(event),
evidence=f"NodeReal风险链上事件: {label}",
value={"value_usd": event.get("value_usd"), "confidence": event.get("confidence")},
)
score_delta -= delta
signals.append(f"⚠️ 链上风险: {label}(置信{event.get('confidence') or 0}, ${float(event.get('value_usd') or 0):.0f})")
if onchain_score >= 75 and not positive_events:
delta = factor_scorer.delta(
"smart_money_buying",
1.5,
evidence="NodeReal综合链上重要性分>=75",
value=onchain_score,
)
score_delta += delta
signals.append(f"链上综合重要性高({onchain_score:.0f})")
if risk_score >= 50 and not risk_events:
delta = factor_scorer.delta(
"holder_concentration_risk",
1.5,
evidence="NodeReal综合链上风险分>=50",
value=risk_score,
)
score_delta -= delta
signals.append(f"⚠️ 链上综合风险高({risk_score:.0f})")
ctx["score_delta"] = round(score_delta, 3)
return score_delta, signals, ctx
# ==================== 确认逻辑 ====================
def detect_volume_price_fly_1h(df_1h):
@ -702,6 +785,7 @@ def confirm_burst(symbol, cand):
signals = []
confirmed = False
entry_plan = {}
factor_scorer = FactorScorer.from_runtime()
# 提取cand数据（v1.7.0：用于辅助信号检测）
cand_detail = json.loads(cand.get("detail_json", "{}"))
@ -723,13 +807,53 @@ def confirm_burst(symbol, cand):
if h1_df is None or len(h1_df) < 50:
return {"confirmed": False, "score": 0, "signals": ["数据不足"], "entry_plan": {},
"pa_1h": {}, "pa_15min": {}, "pa_1d": {}, "m30_aligned": False,
"market_context": {}, "derivatives_context": {}, "sector_context": {}}
"market_context": {}, "derivatives_context": {}, "sector_context": {},
"factor_score_breakdown": {"total_delta": 0, "items": []}}
price = float(h1_df["close"].iloc[-1])
atr_1h = calc_atr(h1_df, 14)
confirm_cfg = _get_cfg_section("confirm")
pa_recency_cfg = confirm_cfg.get("pa_recency", {})
upstream_sector_context = cand_detail.get("sector_context") or {}
if upstream_sector_context.get("hot_sectors") or upstream_sector_context.get("leader_symbol"):
signals.append(
"板块联动: {}{}".format(
",".join(upstream_sector_context.get("hot_sectors") or []),
f" 龙头{upstream_sector_context.get('leader_symbol')}" if upstream_sector_context.get("leader_symbol") else "",
)
)
score += factor_scorer.delta(
"sector_rotation",
2,
evidence="粗筛/细筛板块热度共振",
value=upstream_sector_context,
)
if float(cand_detail.get("sentiment_bonus") or 0) > 0:
signals.append(f"舆情共振(+{cand_detail.get('sentiment_bonus')})")
score += factor_scorer.delta(
"sentiment_resonance",
min(2, float(cand_detail.get("sentiment_bonus") or 0)),
evidence="上游舆情热度进入候选",
value=cand_detail.get("sentiment_bonus"),
)
upstream_deriv = cand_detail.get("derivatives_context") or {}
top_long = upstream_deriv.get("top_trader_long_pct")
if top_long is not None and float(top_long or 0) > 55:
signals.append(f"大户偏多({float(top_long):.0f}%)")
score += factor_scorer.delta(
"top_trader_long",
1,
evidence="Binance futures top trader long pct > 55%",
value=top_long,
)
onchain_delta, onchain_signals, onchain_context = _apply_onchain_factor_score(symbol, factor_scorer)
if onchain_signals:
signals.extend(onchain_signals)
score += onchain_delta
else:
onchain_context = onchain_context or {"has_data": False}
# ---- 1H量价行为（核心前瞻信号） ----
vol_avg = float(h1_df["volume"].rolling(20).mean().iloc[-1])
vol_latest = float(h1_df["volume"].iloc[-1])
@ -746,10 +870,10 @@ def confirm_burst(symbol, cand):
# 量价齐飞K≥2 → 极强确认
if vp_fly_count >= 2:
signals.append(f"1H {vp_fly_count}根量价齐飞K(最强确认)")
score += 8
score += factor_scorer.delta("vp_fly_1h_current", 8, evidence="1H量价齐飞K>=2", value=vp_fly_count)
elif vp_fly_count == 1:
signals.append(f"1H 量价齐飞K(量{vp_data['max_vol_ratio']}x)")
score += 5
score += factor_scorer.delta("vp_fly_1h_current", 5, evidence="1H量价齐飞K=1", value=vp_data.get("max_vol_ratio"))
elif vp_data.get("stale_vp_fly_count", 0) > 0:
stale = vp_data.get("stale_vp_fly_details", [{}])[0]
signals.append(f"1H历史放量阳线已过期({stale.get('age_hours')}小时前, 量{stale.get('vol_ratio')}x)")
@ -757,7 +881,7 @@ def confirm_burst(symbol, cand):
# 1H放量≥3x（但不是量价齐飞=量价背离）
if vol_ratio >= 3 and vp_fly_count == 0:
signals.append(f"1H放量({vol_ratio:.1f}x)但无量价齐飞(量价背离)")
score += 1 # 低权重：量价背离是假信号
score += factor_scorer.delta("volume_divergence_1h", 1, evidence="1H放量但价格行为未确认", value=round(vol_ratio, 2))
# ---- PA引擎：4H级别（阻力/支撑） ----
pa_4h = full_pa_analysis(h4_df, "4h") if h4_df is not None and len(h4_df) >= 30 else {}
@ -782,10 +906,10 @@ def confirm_burst(symbol, cand):
for ig in recent_d1_ignitions[-3:]:
if ig["direction"] == 1:
signals.append(f"日线 {ig['signal_type']}(强度{ig['strength_ratio']}×)")
score += 6 # 日线×1.5 vs 4H的+4
score += factor_scorer.delta("ignition_d1_current", 6, evidence="日线当前多头起爆点", value=ig.get("strength_ratio"))
elif ig["direction"] == -1:
signals.append(f"日线 {ig['signal_type']}(强度{ig['strength_ratio']}×)")
score += 3
score += factor_scorer.delta("trend_exhaustion", 3, evidence="日线空头起爆风险", value=ig.get("strength_ratio"))
for ig in recent_d1_ignitions:
if ig.get("direction") == 1:
t = _event_time_from_age(d1_df, ig.get("age_bars"))
@ -801,10 +925,10 @@ def confirm_burst(symbol, cand):
dy_d1 = sum(1 for c in recent_d1 if c["type"] == "dynamic" and c["direction"] == 1)
if dy_d1 >= 3:
signals.append(f"日线 {dy_d1}动K(阳)趋势确认")
score += 5
score += factor_scorer.delta("dynamic_k_d1_bull", 5, evidence="日线阳动K>=3", value=dy_d1)
elif dy_d1 >= 1:
signals.append(f"日线 {dy_d1}动K(阳)")
score += 2
score += factor_scorer.delta("dynamic_k_d1_bull", 2, evidence="日线阳动K>=1", value=dy_d1)
# 日线需求区反弹 — 最强结构信号
d1_demand = [z for z in d1_zones if z["type"] == "demand" and z["q_score"] >= 5]
@ -812,7 +936,7 @@ def confirm_burst(symbol, cand):
nearest = min(d1_demand, key=lambda z: abs(z["top"] - price))
if nearest["top"] < price < nearest["top"] * 1.15:
signals.append(f"日线需求区反弹(Q={nearest['q_score']} ${nearest['top']:.4f})")
score += 6
score += factor_scorer.delta("breakout_pullback_d1", 6, evidence="日线需求区附近反弹", value=nearest.get("q_score"))
# ---- 1H放量突破4H高质量阻力 ----
breakout_confirmed = False
@ -825,7 +949,7 @@ def confirm_burst(symbol, cand):
elif vol_ratio >= 5:
# 5x以上放量即有效突破（山寨币历史数据不足以形成阻力区）
signals.append(f"1H极放量({vol_ratio:.1f}x)")
score += 2
score += factor_scorer.delta("vp_fly_1h_current", 2, evidence="1H极放量兜底", value=round(vol_ratio, 2))
# ---- PA引擎：1H级别分析 ----
pa_1h = full_pa_analysis(h1_df, "1h") if atr_1h > 0 else {}
@ -840,11 +964,11 @@ def confirm_burst(symbol, cand):
for ig in recent_1h_ignitions[-3:]:
if ig["direction"] == 1:
signals.append(f"1H {ig['signal_type']}(强度{ig['strength_ratio']}×)")
score += 4
score += factor_scorer.delta("ignition_1h_current", 4, evidence="1H当前多头起爆点", value=ig.get("strength_ratio"))
ignition_confirmed = True
elif ig["direction"] == -1:
signals.append(f"1H {ig['signal_type']}(强度{ig['strength_ratio']}×)")
score += 2
score += factor_scorer.delta("trend_exhaustion", 2, evidence="1H空头起爆风险", value=ig.get("strength_ratio"))
for ig in recent_1h_ignitions:
if ig.get("direction") == 1:
t = _event_time_from_age(h1_df, ig.get("age_bars"))
@ -867,10 +991,10 @@ def confirm_burst(symbol, cand):
break
if vol_increasing:
signals.append(f"1H {dy_1h}动K(阳)+量递增")
score += 3
score += factor_scorer.delta("dynamic_k_1h_bull", 3, evidence="1H阳动K且量递增", value=dy_1h)
else:
signals.append(f"1H {dy_1h}动K(阳)")
score += 1
score += factor_scorer.delta("dynamic_k_1h_bull", 1, evidence="1H阳动K", value=dy_1h)
# ---- 1H趋势衰减检测 ----
pa_1h_exhaustion = pa_1h.get("trend_exhaustion", {})
@ -878,9 +1002,9 @@ def confirm_burst(symbol, cand):
for es in pa_1h_exhaustion.get("signals", []):
signals.append(f"⚠️ {es}")
if pa_1h_exhaustion["severity"] == "high":
score -= 3
score -= factor_scorer.delta("trend_exhaustion", 3, evidence="1H高强度趋势衰竭", value=pa_1h_exhaustion.get("severity"))
elif pa_1h_exhaustion["severity"] == "medium":
score -= 1
score -= factor_scorer.delta("trend_exhaustion", 1, evidence="1H中等趋势衰竭", value=pa_1h_exhaustion.get("severity"))
# ---- v1.7.7: 30min 桥接（填补 1H→15min 缺口）----
# 30min 是中间周期：确认 1H 趋势在中等周期是否有结构支撑
@ -896,7 +1020,7 @@ def confirm_burst(symbol, cand):
# 30min 与 1H 方向对齐：阳动K≥3 且 阴动K≤1
if dy_30 >= 3 and sum(1 for c in recent_30 if c["type"] == "dynamic" and c["direction"] == -1) <= 1:
signals.append(f"30min {dy_30}阳动K(与1H共振)")
score += 3
score += factor_scorer.delta("dynamic_k_1h_bull", 3, evidence="30min与1H多头共振", value=dy_30)
m30_aligned = True
elif st_30 >= 4 and dy_30 >= 1:
# 30min 蓄力中（静K多+少量动K）— 待突破，不加分但也不扣
@ -904,7 +1028,7 @@ def confirm_burst(symbol, cand):
m30_aligned = True # 不扣分，视为中性偏多
elif sum(1 for c in recent_30 if c["type"] == "dynamic" and c["direction"] == -1) >= 3:
signals.append("⚠️ 30min 阴动K≥3(与1H背离)")
score -= 2
score -= factor_scorer.delta("trend_exhaustion", 2, evidence="30min阴动K与1H背离", value="30m_bear_dynamic")
# else: 无明确信号，不干预
# ---- PA引擎：15min入场点分析 ----
@ -929,23 +1053,23 @@ def confirm_burst(symbol, cand):
atr_ratio = bk.get("atr_ratio", 0)
if atr_ratio > 2.0:
signals.append(f"15min 强突破K线(ATR×{atr_ratio:.1f})")
score += 3
score += factor_scorer.delta("breakout_15m_current", 3, evidence="15min强突破K", value=round(atr_ratio, 2))
elif atr_ratio > 1.5:
signals.append(f"15min 突破K线(ATR×{atr_ratio:.1f})")
score += 2
score += factor_scorer.delta("breakout_15m_current", 2, evidence="15min突破K", value=round(atr_ratio, 2))
if pa_15min_result.get("pullback_info"):
pb = pa_15min_result["pullback_info"]
signals.append(f"15min 回踩确认(${pb.get('low', 0):.4f}→${pb.get('high', 0):.4f})")
score += 2
score += factor_scorer.delta("pullback_15m_confirm", 2, evidence="15min回踩后承接", value=pb)
if pa_15min_result.get("false_breakout"):
signals.append("⚠️ 15min假突破！排除")
score -= 5
score -= factor_scorer.delta("false_breakout", 5, evidence="15min假突破", value=True)
if entry_action == "即刻买入":
signals.append("🟢 15min即刻入场信号")
score += 3
score += factor_scorer.delta("breakout_15m_current", 3, evidence="15min即刻入场信号", value=entry_action)
elif entry_action == "等回踩":
wait_price = pa_15min_result.get("wait_price", 0)
if wait_price > 0:
@ -965,7 +1089,7 @@ def confirm_burst(symbol, cand):
# ---- 底部突破回踩加分（必须在最终确认判定前生效）----
if bp_daily.get("detected"):
signals.extend(bp_daily.get("signals", []))
score += min(bp_daily["score"], 12)
score += factor_scorer.add_existing("breakout_pullback_d1", bp_daily.get("score", 0), evidence="日线底部突破回踩模型", value=bp_daily, cap=12)
# ---- 最终确认判定（v1.7.0：双门控 — 量价齐飞 OR 强共振旁路）----
# 门控A：量价齐飞K ≥1（保留，历史最可靠）
@ -1050,7 +1174,7 @@ def confirm_burst(symbol, cand):
)
bypass_confirmed = True
confirmed = True
score += bonus
score += factor_scorer.delta("strong_resonance_bypass", bonus, evidence="起爆强度+静K蓄力+辅助信号", value={"max_ig_strength": max_ig_strength, "static_k_count": static_k_count, "aux_count": aux_count})
# ---- 日线趋势安全检查（v1.6.3）----
# 日线持续走低 → 不确认，即使有1H量价齐飞
@ -1098,7 +1222,7 @@ def confirm_burst(symbol, cand):
pullback_cfg = confirm_cfg.get("pullback_penalty", {})
if pullback_cfg.get("enabled", True) and entry_action == "等回踩":
penalty = pullback_cfg.get("score_deduction", 3)
score -= penalty
score -= factor_scorer.delta("risk_reward_bad", penalty, evidence="等回踩历史失败率较高，降低确认分", value=entry_action)
signals.append(f"⚠️ 等回踩降权(-{penalty}分)")
confirmed = confirmed and score >= confirm_min_score()
# 假突破排除
@ -1247,7 +1371,7 @@ def confirm_burst(symbol, cand):
if gate_reasons:
signals.append("⚠️ 买点质量闸门: " + "；".join(gate_reasons[:3]))
if gated_action == "观察":
score -= 2
score -= factor_scorer.delta("entry_quality_gate", 2, evidence="买点质量闸门降为观察", value=gate_reasons[:3])
# 周线突破回踩（需独立拉取）
bp_weekly = {"detected": False}
@ -1260,12 +1384,13 @@ def confirm_burst(symbol, cand):
if bp_weekly.get("detected"):
signals.extend(bp_weekly.get("signals", []))
score += min(bp_weekly["score"], 10)
score += factor_scorer.add_existing("breakout_pullback_w1", bp_weekly.get("score", 0), evidence="周线突破回踩模型", value=bp_weekly, cap=10)
# ---- 计算上下文数据 ----
market_context = compute_market_context(h1_df, price)
derivatives_context = fetch_derivatives_context(symbol)
sector_context = compute_sector_context(symbol, cand_detail)
factor_score_breakdown = factor_scorer.summary()
trigger_context = _build_trigger_context(
fresh_reason if 'fresh_reason' in locals() else "",
fresh_events if 'fresh_events' in locals() else [],
@ -1277,10 +1402,15 @@ def confirm_burst(symbol, cand):
entry_action=entry_action,
)
market_context["trigger_context"] = trigger_context
market_context["factor_score_breakdown"] = factor_score_breakdown
market_context["onchain_context"] = onchain_context
if entry_plan:
entry_plan["factor_score_breakdown"] = factor_score_breakdown
entry_plan["onchain_context"] = onchain_context
return {
"confirmed": confirmed,
"score": score,
"score": round(float(score), 3),
"signals": signals,
"entry_plan": entry_plan if confirmed else {},
"price": round(float(price), 6),
@ -1292,6 +1422,8 @@ def confirm_burst(symbol, cand):
"market_context": market_context,
"derivatives_context": derivatives_context,
"sector_context": sector_context,
"onchain_context": onchain_context,
"factor_score_breakdown": factor_score_breakdown,
"fresh_reason": fresh_reason if 'fresh_reason' in locals() else "",
"fresh_events": fresh_events if 'fresh_events' in locals() else [],
"trigger_context": trigger_context if 'trigger_context' in locals() else {},

30
tests/test_factor_scoring.py Normal file
View File

@ -0,0 +1,30 @@
from app.config.config_loader import get_signal_weights
from app.core.factor_scoring import FactorScorer
def test_factor_scorer_eliminates_zero_weight_factor():
scorer = FactorScorer(weights={"vp_fly_1h_current": 0})
delta = scorer.delta("vp_fly_1h_current", 8, evidence="unit")
assert delta == 0
assert scorer.summary()["items"][0]["source"] == "review_weight"
def test_factor_scorer_promotes_reviewed_factor():
scorer = FactorScorer(weights={"vp_fly_1h_current": 7.5})
delta = scorer.delta("vp_fly_1h_current", 5, evidence="unit")
assert delta == 7.5
assert scorer.summary()["items"][0]["multiplier"] == 1.5
def test_signal_weight_alias_keeps_legacy_chinese_keys_available(monkeypatch):
monkeypatch.setattr("app.config.config_loader.load_rules", lambda: {"signal_weights": {"量价齐飞": 5}})
monkeypatch.setattr("app.db.altcoin_db.get_signal_weights", lambda: {})
weights = get_signal_weights()
assert weights["vp_fly_1h_current"] == 5
assert weights["量价齐飞"] == 5

79
tests/test_onchain_factor_scoring.py Normal file
View File

@ -0,0 +1,79 @@
from datetime import datetime
import pytest
from app.core.factor_scoring import FactorScorer
from app.db.onchain_db import get_onchain_factor_context, insert_onchain_event, insert_token_metric
from app.services.altcoin_confirm import _apply_onchain_factor_score
def test_onchain_factor_context_returns_recent_positive_and_risk_events():
now = datetime.now().isoformat()
insert_token_metric(
{
"symbol": "BEAM/USDT",
"chain": "ethereum",
"window": "1h",
"metric_time": now,
"onchain_score": 82,
"risk_score": 12,
"source": "nodereal",
}
)
insert_onchain_event(
{
"symbol": "BEAM/USDT",
"chain": "ethereum",
"signal_code": "whale_accumulation",
"direction": "positive",
"value_usd": 1_200_000,
"confidence": 88,
"detected_at": now,
"source": "nodereal",
}
)
insert_onchain_event(
{
"symbol": "BEAM/USDT",
"chain": "ethereum",
"signal_code": "exchange_inflow_risk",
"direction": "risk",
"value_usd": 800_000,
"confidence": 80,
"detected_at": now,
"source": "nodereal",
}
)
ctx = get_onchain_factor_context("BEAM/USDT", hours=24)
assert ctx["has_data"] is True
assert ctx["metrics"]["onchain_score"] == pytest.approx(82)
assert ctx["positive_event_count"] == 1
assert ctx["risk_event_count"] == 1
assert ctx["top_positive"]["signal_code"] == "whale_accumulation"
def test_onchain_factor_score_uses_review_weights():
now = datetime.now().isoformat()
insert_onchain_event(
{
"symbol": "SMART/USDT",
"chain": "bsc",
"signal_code": "smart_money_buying",
"direction": "positive",
"value_usd": 500_000,
"confidence": 76,
"detected_at": now,
"source": "nodereal",
}
)
scorer = FactorScorer(weights={"smart_money_buying": 0})
delta, signals, ctx = _apply_onchain_factor_score("SMART/USDT", scorer)
assert ctx["has_data"] is True
assert signals
assert delta == 0
assert scorer.summary()["items"][0]["factor_code"] == "smart_money_buying"
assert scorer.summary()["items"][0]["score_delta"] == 0