feat: AI 即时分析计费/共识/校准与 Docker 前端构建
- 即时分析：先扣费、防重入(429)、失败退款；记忆库与离线校准 worker - 多周期共识、客观分与设置项 AI_ANALYSIS_CONSENSUS_TIMEFRAMES - Docker：前端多阶段构建(QuantDinger-Vue-src)、根目录 .dockerignore、compose 调整 - 同步 frontend/dist 静态资源 Made-with: Cursor
This commit is contained in:
dinger
@@ -273,6 +273,12 @@ def create_app(config_name='default'):
|
||||
start_portfolio_monitor()
|
||||
start_usdt_order_worker()
|
||||
start_polymarket_worker()
|
||||
# Offline calibration to make AI thresholds self-tuning.
|
||||
try:
|
||||
from app.services.ai_calibration import start_ai_calibration_worker
|
||||
start_ai_calibration_worker()
|
||||
except Exception:
|
||||
pass
|
||||
restore_running_strategies()
|
||||
|
||||
return app
|
||||
|
||||
@@ -4,16 +4,45 @@ Fast Analysis API Routes
|
||||
New high-performance analysis endpoints that replace the slow multi-agent system.
|
||||
"""
|
||||
from flask import Blueprint, request, jsonify, g
|
||||
import threading
|
||||
import time
|
||||
|
||||
from app.utils.auth import login_required
|
||||
from app.utils.logger import get_logger
|
||||
from app.services.fast_analysis import get_fast_analysis_service
|
||||
from app.services.analysis_memory import get_analysis_memory
|
||||
from app.services.billing_service import get_billing_service
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
fast_analysis_bp = Blueprint('fast_analysis', __name__)
|
||||
|
||||
# In-memory in-flight guard to avoid duplicate analysis charges caused by rapid repeated clicks.
|
||||
_analysis_inflight_lock = threading.Lock()
|
||||
_analysis_inflight = {} # key -> expire_ts
|
||||
|
||||
|
||||
def _build_inflight_key(user_id: int, market: str, symbol: str, timeframe: str) -> str:
|
||||
return f"{int(user_id)}|{str(market or '').strip().upper()}|{str(symbol or '').strip().upper()}|{str(timeframe or '').strip().upper()}"
|
||||
|
||||
|
||||
def _acquire_inflight(key: str, ttl_sec: int = 90) -> bool:
|
||||
now = time.time()
|
||||
with _analysis_inflight_lock:
|
||||
# Cleanup stale entries
|
||||
stale = [k for k, exp in _analysis_inflight.items() if float(exp) <= now]
|
||||
for k in stale[:1024]:
|
||||
_analysis_inflight.pop(k, None)
|
||||
if key in _analysis_inflight and float(_analysis_inflight.get(key) or 0) > now:
|
||||
return False
|
||||
_analysis_inflight[key] = now + int(ttl_sec)
|
||||
return True
|
||||
|
||||
|
||||
def _release_inflight(key: str):
|
||||
with _analysis_inflight_lock:
|
||||
_analysis_inflight.pop(key, None)
|
||||
|
||||
|
||||
@fast_analysis_bp.route('/analyze', methods=['POST'])
|
||||
@login_required
|
||||
@@ -51,6 +80,58 @@ def analyze():
|
||||
|
||||
# Get current user's ID to associate analysis with user
|
||||
user_id = getattr(g, 'user_id', None)
|
||||
if not user_id:
|
||||
return jsonify({'code': 0, 'msg': 'Unauthorized', 'data': None}), 401
|
||||
|
||||
inflight_key = _build_inflight_key(user_id, market, symbol, timeframe)
|
||||
if not _acquire_inflight(inflight_key, ttl_sec=90):
|
||||
return jsonify({
|
||||
'code': 0,
|
||||
'msg': 'Analysis already in progress for this symbol/timeframe. Please wait.',
|
||||
'data': {'in_progress': True}
|
||||
}), 429
|
||||
|
||||
# Billing / credits (best-effort, consistent with polymarket deep analysis)
|
||||
credits_charged = 0
|
||||
remaining_credits = None
|
||||
billing_consumed = False
|
||||
billing = None
|
||||
try:
|
||||
billing = get_billing_service()
|
||||
if billing.is_billing_enabled():
|
||||
credits_charged = int(billing.get_feature_cost('ai_analysis') or 0)
|
||||
if credits_charged > 0:
|
||||
ok, msg = billing.check_and_consume(
|
||||
user_id=int(user_id),
|
||||
feature='ai_analysis',
|
||||
reference_id=f"fast_analysis_{market}:{symbol}:{timeframe}"
|
||||
)
|
||||
if not ok:
|
||||
# Standardize insufficient credits message
|
||||
if str(msg or "").startswith('insufficient_credits'):
|
||||
# Format: insufficient_credits:<current>:<cost>
|
||||
parts = str(msg).split(':')
|
||||
cur = float(parts[1]) if len(parts) >= 2 else 0.0
|
||||
req = float(parts[2]) if len(parts) >= 3 else float(credits_charged)
|
||||
return jsonify({
|
||||
'code': 0,
|
||||
'msg': 'Insufficient credits',
|
||||
'data': {
|
||||
'required': req,
|
||||
'current': cur,
|
||||
'shortage': max(0.0, req - cur),
|
||||
}
|
||||
}), 400
|
||||
return jsonify({'code': 0, 'msg': f'Failed to deduct credits: {msg}', 'data': None}), 500
|
||||
billing_consumed = True
|
||||
# Query remaining credits after successful consumption
|
||||
try:
|
||||
remaining_credits = float(billing.get_user_credits(int(user_id)))
|
||||
except Exception:
|
||||
remaining_credits = None
|
||||
except Exception as e:
|
||||
# Billing failure should not crash analysis by default, but should be visible in logs.
|
||||
logger.warning(f"Billing check failed (skipped): {e}", exc_info=True)
|
||||
|
||||
service = get_fast_analysis_service()
|
||||
result = service.analyze(
|
||||
@@ -63,6 +144,18 @@ def analyze():
|
||||
)
|
||||
|
||||
if result.get('error'):
|
||||
# Best-effort refund if we already charged but analysis failed.
|
||||
if billing_consumed and billing and credits_charged > 0:
|
||||
try:
|
||||
billing.add_credits(
|
||||
user_id=int(user_id),
|
||||
amount=int(credits_charged),
|
||||
action='refund',
|
||||
remark=f'Auto refund: fast-analysis failed ({market}:{symbol}:{timeframe})'
|
||||
)
|
||||
remaining_credits = float(billing.get_user_credits(int(user_id)))
|
||||
except Exception as re:
|
||||
logger.error(f"Auto refund failed: {re}", exc_info=True)
|
||||
return jsonify({
|
||||
'code': 0,
|
||||
'msg': result['error'],
|
||||
@@ -75,16 +168,37 @@ def analyze():
|
||||
return jsonify({
|
||||
'code': 1,
|
||||
'msg': 'success',
|
||||
'data': result
|
||||
'data': {
|
||||
**(result or {}),
|
||||
'credits_charged': credits_charged,
|
||||
'remaining_credits': remaining_credits,
|
||||
}
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
# Best-effort refund on unexpected error after charge.
|
||||
try:
|
||||
if 'billing_consumed' in locals() and billing_consumed and 'billing' in locals() and billing and credits_charged > 0 and 'user_id' in locals() and user_id:
|
||||
billing.add_credits(
|
||||
user_id=int(user_id),
|
||||
amount=int(credits_charged),
|
||||
action='refund',
|
||||
remark=f'Auto refund: fast-analysis exception ({market}:{symbol}:{timeframe})'
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
logger.error(f"Fast analysis API failed: {e}", exc_info=True)
|
||||
return jsonify({
|
||||
'code': 0,
|
||||
'msg': str(e),
|
||||
'data': None
|
||||
}), 500
|
||||
finally:
|
||||
try:
|
||||
if 'inflight_key' in locals() and inflight_key:
|
||||
_release_inflight(inflight_key)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
@fast_analysis_bp.route('/analyze-legacy', methods=['POST'])
|
||||
@@ -115,6 +229,56 @@ def analyze_legacy():
|
||||
'msg': 'market and symbol are required',
|
||||
'data': None
|
||||
}), 400
|
||||
|
||||
# Billing / credits (same behavior as /analyze)
|
||||
user_id = getattr(g, 'user_id', None)
|
||||
if not user_id:
|
||||
return jsonify({'code': 0, 'msg': 'Unauthorized', 'data': None}), 401
|
||||
|
||||
inflight_key = _build_inflight_key(user_id, market, symbol, timeframe)
|
||||
if not _acquire_inflight(inflight_key, ttl_sec=90):
|
||||
return jsonify({
|
||||
'code': 0,
|
||||
'msg': 'Analysis already in progress for this symbol/timeframe. Please wait.',
|
||||
'data': {'in_progress': True}
|
||||
}), 429
|
||||
|
||||
credits_charged = 0
|
||||
remaining_credits = None
|
||||
billing_consumed = False
|
||||
billing = None
|
||||
try:
|
||||
billing = get_billing_service()
|
||||
if billing.is_billing_enabled():
|
||||
credits_charged = int(billing.get_feature_cost('ai_analysis') or 0)
|
||||
if credits_charged > 0:
|
||||
ok, msg = billing.check_and_consume(
|
||||
user_id=int(user_id),
|
||||
feature='ai_analysis',
|
||||
reference_id=f"fast_analysis_legacy_{market}:{symbol}:{timeframe}"
|
||||
)
|
||||
if not ok:
|
||||
if str(msg or "").startswith('insufficient_credits'):
|
||||
parts = str(msg).split(':')
|
||||
cur = float(parts[1]) if len(parts) >= 2 else 0.0
|
||||
req = float(parts[2]) if len(parts) >= 3 else float(credits_charged)
|
||||
return jsonify({
|
||||
'code': 0,
|
||||
'msg': 'Insufficient credits',
|
||||
'data': {
|
||||
'required': req,
|
||||
'current': cur,
|
||||
'shortage': max(0.0, req - cur),
|
||||
}
|
||||
}), 400
|
||||
return jsonify({'code': 0, 'msg': f'Failed to deduct credits: {msg}', 'data': None}), 500
|
||||
billing_consumed = True
|
||||
try:
|
||||
remaining_credits = float(billing.get_user_credits(int(user_id)))
|
||||
except Exception:
|
||||
remaining_credits = None
|
||||
except Exception as e:
|
||||
logger.warning(f"Billing check failed (skipped): {e}", exc_info=True)
|
||||
|
||||
service = get_fast_analysis_service()
|
||||
result = service.analyze_legacy_format(
|
||||
@@ -126,6 +290,17 @@ def analyze_legacy():
|
||||
)
|
||||
|
||||
if result.get('error'):
|
||||
if billing_consumed and billing and credits_charged > 0:
|
||||
try:
|
||||
billing.add_credits(
|
||||
user_id=int(user_id),
|
||||
amount=int(credits_charged),
|
||||
action='refund',
|
||||
remark=f'Auto refund: fast-analysis-legacy failed ({market}:{symbol}:{timeframe})'
|
||||
)
|
||||
remaining_credits = float(billing.get_user_credits(int(user_id)))
|
||||
except Exception as re:
|
||||
logger.error(f"Auto refund failed (legacy): {re}", exc_info=True)
|
||||
return jsonify({
|
||||
'code': 0,
|
||||
'msg': result['error'],
|
||||
@@ -135,16 +310,36 @@ def analyze_legacy():
|
||||
return jsonify({
|
||||
'code': 1,
|
||||
'msg': 'success',
|
||||
'data': result
|
||||
'data': {
|
||||
**(result or {}),
|
||||
'credits_charged': credits_charged,
|
||||
'remaining_credits': remaining_credits,
|
||||
}
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
try:
|
||||
if 'billing_consumed' in locals() and billing_consumed and 'billing' in locals() and billing and credits_charged > 0 and 'user_id' in locals() and user_id:
|
||||
billing.add_credits(
|
||||
user_id=int(user_id),
|
||||
amount=int(credits_charged),
|
||||
action='refund',
|
||||
remark=f'Auto refund: fast-analysis-legacy exception ({market}:{symbol}:{timeframe})'
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
logger.error(f"Fast analysis legacy API failed: {e}", exc_info=True)
|
||||
return jsonify({
|
||||
'code': 0,
|
||||
'msg': str(e),
|
||||
'data': None
|
||||
}), 500
|
||||
finally:
|
||||
try:
|
||||
if 'inflight_key' in locals() and inflight_key:
|
||||
_release_inflight(inflight_key)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
@fast_analysis_bp.route('/history', methods=['GET'])
|
||||
|
||||
@@ -220,6 +220,14 @@ CONFIG_SCHEMA = {
|
||||
'required': False,
|
||||
'description': 'Custom model list in JSON format for model selector'
|
||||
},
|
||||
{
|
||||
'key': 'AI_ANALYSIS_CONSENSUS_TIMEFRAMES',
|
||||
'label': 'Consensus Timeframes',
|
||||
'type': 'text',
|
||||
'default': '1D,4H',
|
||||
'required': False,
|
||||
'description': 'Multi-timeframe consensus for fast AI analysis. Comma-separated, e.g. "1D,4H"'
|
||||
},
|
||||
]
|
||||
},
|
||||
|
||||
|
||||
@@ -0,0 +1,348 @@
|
||||
"""
|
||||
AI Calibration Service (offline).
|
||||
|
||||
Goal:
|
||||
- Calibrate the objective-score -> decision thresholds using validated historical analysis outcomes.
|
||||
- Make FastAnalysisService "self-tuning" based on performance.
|
||||
|
||||
Approach (approx rules as requested):
|
||||
- Use qd_analysis_memory.actual_return_pct and apply simple correctness rules:
|
||||
BUY correct if return_pct > +2
|
||||
SELL correct if return_pct < -2
|
||||
HOLD correct if abs(return_pct) <= 5
|
||||
- Search candidate absolute thresholds for score mapping:
|
||||
score >= +thr => BUY
|
||||
score <= -thr => SELL
|
||||
else => HOLD
|
||||
|
||||
We calibrate on consensus_score because it's the main "objective" signal used for overriding decisions.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Dict, Any, List, Optional, Tuple
|
||||
|
||||
from app.utils.db import get_db_connection
|
||||
from app.utils.logger import get_logger
|
||||
from app.services.analysis_memory import get_analysis_memory, AnalysisMemory
|
||||
from app.services.market_data_collector import MarketDataCollector
|
||||
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
DEFAULTS = {
|
||||
"buy_threshold": 20.0,
|
||||
"sell_threshold": -20.0,
|
||||
"min_consensus_abs_override": 15.0,
|
||||
"quality_hold_threshold": 0.7,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class CalibrationResult:
|
||||
market: str
|
||||
buy_threshold: float
|
||||
sell_threshold: float
|
||||
best_accuracy: float
|
||||
coverage: Dict[str, int]
|
||||
sample_count: int
|
||||
validated_count: int
|
||||
updated_at_ts: float
|
||||
|
||||
|
||||
class AICalibrationService:
|
||||
def __init__(self):
|
||||
self._ensure_table()
|
||||
|
||||
def _ensure_table(self) -> None:
|
||||
try:
|
||||
with get_db_connection() as db:
|
||||
cur = db.cursor()
|
||||
cur.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS qd_ai_calibration (
|
||||
id SERIAL PRIMARY KEY,
|
||||
market VARCHAR(50) NOT NULL,
|
||||
buy_threshold DECIMAL(10,4) NOT NULL,
|
||||
sell_threshold DECIMAL(10,4) NOT NULL,
|
||||
min_consensus_abs_override DECIMAL(10,4) NOT NULL,
|
||||
quality_hold_threshold DECIMAL(10,4) NOT NULL,
|
||||
sample_count INT NOT NULL DEFAULT 0,
|
||||
validated_at TIMESTAMP DEFAULT NOW(),
|
||||
created_at TIMESTAMP DEFAULT NOW()
|
||||
);
|
||||
"""
|
||||
)
|
||||
# Index for latest lookup
|
||||
cur.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_ai_calibration_market_validated_at
|
||||
ON qd_ai_calibration(market, validated_at DESC);
|
||||
"""
|
||||
)
|
||||
db.commit()
|
||||
cur.close()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to ensure qd_ai_calibration table: {e}", exc_info=True)
|
||||
|
||||
def get_latest(self, market: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Get latest calibration config for market.
|
||||
Falls back to DEFAULTS if not found.
|
||||
"""
|
||||
market = (market or "").strip()
|
||||
if not market:
|
||||
return dict(DEFAULTS)
|
||||
|
||||
try:
|
||||
with get_db_connection() as db:
|
||||
cur = db.cursor()
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT buy_threshold, sell_threshold,
|
||||
min_consensus_abs_override, quality_hold_threshold
|
||||
FROM qd_ai_calibration
|
||||
WHERE market = %s
|
||||
ORDER BY validated_at DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(market,),
|
||||
)
|
||||
row = cur.fetchone() or {}
|
||||
cur.close()
|
||||
if not row:
|
||||
return dict(DEFAULTS)
|
||||
out = dict(DEFAULTS)
|
||||
out["buy_threshold"] = float(row.get("buy_threshold") or DEFAULTS["buy_threshold"])
|
||||
out["sell_threshold"] = float(row.get("sell_threshold") or DEFAULTS["sell_threshold"])
|
||||
out["min_consensus_abs_override"] = float(
|
||||
row.get("min_consensus_abs_override") or DEFAULTS["min_consensus_abs_override"]
|
||||
)
|
||||
out["quality_hold_threshold"] = float(
|
||||
row.get("quality_hold_threshold") or DEFAULTS["quality_hold_threshold"]
|
||||
)
|
||||
return out
|
||||
except Exception as e:
|
||||
logger.warning(f"get_latest calibration failed: {e}", exc_info=True)
|
||||
return dict(DEFAULTS)
|
||||
|
||||
def _candidate_abs_thresholds(self) -> List[float]:
|
||||
env = os.getenv("AI_CALIBRATION_CANDIDATE_ABS_THRESHOLDS", "").strip()
|
||||
if env:
|
||||
parts = [p.strip() for p in env.split(",") if p.strip()]
|
||||
out = []
|
||||
for p in parts:
|
||||
try:
|
||||
out.append(float(p))
|
||||
except Exception:
|
||||
continue
|
||||
if out:
|
||||
return sorted(set(out))
|
||||
# Default grid
|
||||
return [10, 12, 14, 16, 18, 20, 22, 25, 30]
|
||||
|
||||
def _correctness_for_return(self, decision: str, return_pct: float) -> bool:
|
||||
# Approx correctness rules (as requested)
|
||||
if decision == "BUY":
|
||||
return return_pct > 2.0
|
||||
if decision == "SELL":
|
||||
return return_pct < -2.0
|
||||
# HOLD
|
||||
return abs(return_pct) <= 5.0
|
||||
|
||||
def _predict_decision_from_score(self, score: float, abs_thr: float) -> str:
|
||||
if score >= abs_thr:
|
||||
return "BUY"
|
||||
if score <= -abs_thr:
|
||||
return "SELL"
|
||||
return "HOLD"
|
||||
|
||||
def calibrate_market(
|
||||
self,
|
||||
market: str = "Crypto",
|
||||
*,
|
||||
lookback_days: int = 30,
|
||||
min_samples: int = 80,
|
||||
validate_before: bool = True,
|
||||
) -> Optional[CalibrationResult]:
|
||||
market = (market or "").strip()
|
||||
if not market:
|
||||
return None
|
||||
|
||||
abs_thresholds = self._candidate_abs_thresholds()
|
||||
|
||||
validated_count = 0
|
||||
try:
|
||||
# Best-effort: validate old unvalidated records first.
|
||||
if validate_before:
|
||||
memory: AnalysisMemory = get_analysis_memory()
|
||||
# Validate anything older than ~7 days (matching your existing approx rules).
|
||||
validated_stats = memory.validate_unvalidated_older_than(
|
||||
min_age_days=7, limit=300
|
||||
)
|
||||
validated_count = int(validated_stats.get("validated", 0) or 0)
|
||||
except Exception as e:
|
||||
logger.warning(f"pre-validation failed (skipped): {e}", exc_info=True)
|
||||
|
||||
# Fetch validated rows with consensus_score and actual_return_pct
|
||||
rows: List[Dict[str, Any]] = []
|
||||
try:
|
||||
with get_db_connection() as db:
|
||||
cur = db.cursor()
|
||||
# Use f-string for interval since Postgres doesn't allow placeholder in INTERVAL literal
|
||||
cur.execute(
|
||||
f"""
|
||||
SELECT
|
||||
decision,
|
||||
consensus_score,
|
||||
consensus_abs,
|
||||
quality_multiplier,
|
||||
agreement_ratio,
|
||||
actual_return_pct
|
||||
FROM qd_analysis_memory
|
||||
WHERE market = %s
|
||||
AND validated_at IS NOT NULL
|
||||
AND actual_return_pct IS NOT NULL
|
||||
AND consensus_score IS NOT NULL
|
||||
AND created_at > NOW() - INTERVAL '{int(lookback_days)} days'
|
||||
""",
|
||||
(market,),
|
||||
)
|
||||
rows = cur.fetchall() or []
|
||||
cur.close()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch memory rows for calibration: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
sample_count = len(rows)
|
||||
if sample_count < min_samples:
|
||||
logger.warning(
|
||||
f"[AI Calibration] Not enough samples for {market}: {sample_count} < min_samples={min_samples}"
|
||||
)
|
||||
return None
|
||||
|
||||
best_abs_thr = abs_thresholds[0]
|
||||
best_accuracy = -1.0
|
||||
best_coverage: Dict[str, int] = {"BUY": 0, "SELL": 0, "HOLD": 0}
|
||||
|
||||
# Evaluate each threshold
|
||||
for thr in abs_thresholds:
|
||||
correct = 0
|
||||
total = 0
|
||||
coverage = {"BUY": 0, "SELL": 0, "HOLD": 0}
|
||||
|
||||
for r in rows:
|
||||
try:
|
||||
score = float(r.get("consensus_score") or 0.0)
|
||||
return_pct = float(r.get("actual_return_pct") or 0.0)
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
pred = self._predict_decision_from_score(score, thr)
|
||||
coverage[pred] += 1
|
||||
total += 1
|
||||
if self._correctness_for_return(pred, return_pct):
|
||||
correct += 1
|
||||
|
||||
if total <= 0:
|
||||
continue
|
||||
acc = correct / total * 100.0
|
||||
|
||||
# Tie-break: prefer higher BUY+SELL coverage (avoid always HOLD)
|
||||
# Secondary tie-break: higher accuracy
|
||||
buy_sell_cov = coverage["BUY"] + coverage["SELL"]
|
||||
best_buy_sell_cov = best_coverage["BUY"] + best_coverage["SELL"]
|
||||
|
||||
if acc > best_accuracy:
|
||||
best_accuracy = acc
|
||||
best_abs_thr = thr
|
||||
best_coverage = coverage
|
||||
elif acc == best_accuracy:
|
||||
if buy_sell_cov > best_buy_sell_cov:
|
||||
best_abs_thr = thr
|
||||
best_coverage = coverage
|
||||
|
||||
# Write new calibration row
|
||||
buy_threshold = float(best_abs_thr)
|
||||
sell_threshold = float(-best_abs_thr)
|
||||
cfg = self.get_latest(market)
|
||||
min_consensus_abs_override = float(cfg.get("min_consensus_abs_override") or DEFAULTS["min_consensus_abs_override"])
|
||||
quality_hold_threshold = float(cfg.get("quality_hold_threshold") or DEFAULTS["quality_hold_threshold"])
|
||||
|
||||
try:
|
||||
with get_db_connection() as db:
|
||||
cur = db.cursor()
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO qd_ai_calibration
|
||||
(market, buy_threshold, sell_threshold,
|
||||
min_consensus_abs_override, quality_hold_threshold,
|
||||
sample_count, validated_at, created_at)
|
||||
VALUES
|
||||
(%s, %s, %s, %s, %s, %s, NOW(), NOW())
|
||||
""",
|
||||
(
|
||||
market,
|
||||
buy_threshold,
|
||||
sell_threshold,
|
||||
min_consensus_abs_override,
|
||||
quality_hold_threshold,
|
||||
sample_count,
|
||||
),
|
||||
)
|
||||
db.commit()
|
||||
cur.close()
|
||||
except Exception as e:
|
||||
logger.error(f"[AI Calibration] Failed to persist calibration: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
return CalibrationResult(
|
||||
market=market,
|
||||
buy_threshold=buy_threshold,
|
||||
sell_threshold=sell_threshold,
|
||||
best_accuracy=float(best_accuracy),
|
||||
coverage=best_coverage,
|
||||
sample_count=sample_count,
|
||||
validated_count=int(validated_count),
|
||||
updated_at_ts=time.time(),
|
||||
)
|
||||
|
||||
|
||||
def start_ai_calibration_worker() -> None:
|
||||
"""
|
||||
Run offline calibration once on service startup (best-effort).
|
||||
"""
|
||||
is_demo_mode = os.getenv("IS_DEMO_MODE", "false").lower() == "true"
|
||||
if is_demo_mode:
|
||||
logger.info("AI calibration worker skipped in demo mode.")
|
||||
return
|
||||
|
||||
enabled = os.getenv("ENABLE_OFFLINE_AI_CALIBRATION", "true").lower() == "true"
|
||||
if not enabled:
|
||||
logger.info("AI calibration worker disabled (ENABLE_OFFLINE_AI_CALIBRATION=false).")
|
||||
return
|
||||
|
||||
try:
|
||||
svc = AICalibrationService()
|
||||
lookback_days = int(os.getenv("AI_CALIBRATION_LOOKBACK_DAYS", "30"))
|
||||
min_samples = int(os.getenv("AI_CALIBRATION_MIN_SAMPLES", "80"))
|
||||
market = os.getenv("AI_CALIBRATION_MARKET", "Crypto").strip() or "Crypto"
|
||||
logger.info(
|
||||
f"Starting offline AI calibration: market={market}, lookback_days={lookback_days}, min_samples={min_samples}"
|
||||
)
|
||||
result = svc.calibrate_market(market=market, lookback_days=lookback_days, min_samples=min_samples)
|
||||
if result:
|
||||
logger.info(
|
||||
f"[AI Calibration] market={result.market} best_thr=+{result.buy_threshold:.1f} "
|
||||
f"accuracy={result.best_accuracy:.2f}% sample={result.sample_count} "
|
||||
f"coverage={result.coverage} validated_new={result.validated_count}"
|
||||
)
|
||||
else:
|
||||
logger.info("[AI Calibration] No calibration update applied (not enough data).")
|
||||
except Exception as e:
|
||||
logger.error(f"start_ai_calibration_worker failed: {e}", exc_info=True)
|
||||
|
||||
@@ -67,6 +67,11 @@ class AnalysisMemory:
|
||||
scores JSONB,
|
||||
indicators_snapshot JSONB,
|
||||
raw_result JSONB,
|
||||
consensus_decision VARCHAR(10),
|
||||
consensus_score DECIMAL(24, 8),
|
||||
consensus_abs DECIMAL(24, 8),
|
||||
agreement_ratio DECIMAL(10, 6),
|
||||
quality_multiplier DECIMAL(10, 6),
|
||||
created_at TIMESTAMP DEFAULT NOW(),
|
||||
validated_at TIMESTAMP,
|
||||
actual_outcome VARCHAR(20),
|
||||
@@ -96,6 +101,42 @@ class AnalysisMemory:
|
||||
) THEN
|
||||
ALTER TABLE qd_analysis_memory ADD COLUMN raw_result JSONB;
|
||||
END IF;
|
||||
|
||||
-- 添加多周期共识列（如果不存在）
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM information_schema.columns
|
||||
WHERE table_name = 'qd_analysis_memory' AND column_name = 'consensus_decision'
|
||||
) THEN
|
||||
ALTER TABLE qd_analysis_memory ADD COLUMN consensus_decision VARCHAR(10);
|
||||
END IF;
|
||||
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM information_schema.columns
|
||||
WHERE table_name = 'qd_analysis_memory' AND column_name = 'consensus_score'
|
||||
) THEN
|
||||
ALTER TABLE qd_analysis_memory ADD COLUMN consensus_score DECIMAL(24, 8);
|
||||
END IF;
|
||||
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM information_schema.columns
|
||||
WHERE table_name = 'qd_analysis_memory' AND column_name = 'consensus_abs'
|
||||
) THEN
|
||||
ALTER TABLE qd_analysis_memory ADD COLUMN consensus_abs DECIMAL(24, 8);
|
||||
END IF;
|
||||
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM information_schema.columns
|
||||
WHERE table_name = 'qd_analysis_memory' AND column_name = 'agreement_ratio'
|
||||
) THEN
|
||||
ALTER TABLE qd_analysis_memory ADD COLUMN agreement_ratio DECIMAL(10, 6);
|
||||
END IF;
|
||||
|
||||
IF NOT EXISTS (
|
||||
SELECT 1 FROM information_schema.columns
|
||||
WHERE table_name = 'qd_analysis_memory' AND column_name = 'quality_multiplier'
|
||||
) THEN
|
||||
ALTER TABLE qd_analysis_memory ADD COLUMN quality_multiplier DECIMAL(10, 6);
|
||||
END IF;
|
||||
END $$;
|
||||
""")
|
||||
|
||||
@@ -150,16 +191,26 @@ class AnalysisMemory:
|
||||
scores = json.dumps(analysis_result.get("scores", {}))
|
||||
indicators = json.dumps(analysis_result.get("indicators", {}))
|
||||
raw = json.dumps(analysis_result)
|
||||
|
||||
consensus = analysis_result.get("consensus") or {}
|
||||
consensus_decision = consensus.get("consensus_decision")
|
||||
consensus_score = consensus.get("consensus_score")
|
||||
consensus_abs = consensus.get("consensus_abs")
|
||||
agreement_ratio = consensus.get("agreement_ratio")
|
||||
quality_multiplier = consensus.get("quality_multiplier")
|
||||
|
||||
cur.execute("""
|
||||
INSERT INTO qd_analysis_memory (
|
||||
user_id, market, symbol, decision, confidence,
|
||||
price_at_analysis, entry_price, stop_loss, take_profit,
|
||||
summary, reasons, risks, scores, indicators_snapshot, raw_result
|
||||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||||
summary, reasons, risks, scores, indicators_snapshot, raw_result,
|
||||
consensus_decision, consensus_score, consensus_abs, agreement_ratio, quality_multiplier
|
||||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
|
||||
%s, %s, %s, %s, %s)
|
||||
RETURNING id
|
||||
""", (user_id, market, symbol, decision, confidence, price, entry, stop, take,
|
||||
summary, reasons, risks, scores, indicators, raw))
|
||||
""", (user_id, market, symbol, decision, confidence, price, entry, stop, take,
|
||||
summary, reasons, risks, scores, indicators, raw,
|
||||
consensus_decision, consensus_score, consensus_abs, agreement_ratio, quality_multiplier))
|
||||
|
||||
# 使用 lastrowid 属性获取 ID（execute 内部已经处理了 RETURNING）
|
||||
memory_id = cur.lastrowid
|
||||
@@ -512,6 +563,84 @@ class AnalysisMemory:
|
||||
|
||||
logger.info(f"Validation completed: {stats}")
|
||||
return stats
|
||||
|
||||
def validate_unvalidated_older_than(self, min_age_days: int = 7, limit: int = 200) -> Dict[str, Any]:
|
||||
"""
|
||||
Best-effort backfill:
|
||||
Validate unvalidated decisions older than `min_age_days`.
|
||||
|
||||
This is used by offline AI calibration so the system can tune itself automatically.
|
||||
"""
|
||||
from app.services.market_data_collector import MarketDataCollector
|
||||
collector = MarketDataCollector()
|
||||
|
||||
stats = {
|
||||
"validated": 0,
|
||||
"correct": 0,
|
||||
"incorrect": 0,
|
||||
"errors": 0,
|
||||
}
|
||||
|
||||
try:
|
||||
with get_db_connection() as db:
|
||||
cur = db.cursor()
|
||||
cur.execute(
|
||||
f"""
|
||||
SELECT id, market, symbol, decision, price_at_analysis
|
||||
FROM qd_analysis_memory
|
||||
WHERE validated_at IS NULL
|
||||
AND created_at < NOW() - INTERVAL '{int(min_age_days)} days'
|
||||
LIMIT {int(limit)}
|
||||
"""
|
||||
)
|
||||
rows = cur.fetchall() or []
|
||||
|
||||
for row in rows:
|
||||
try:
|
||||
current_price = collector._get_price(row["market"], row["symbol"])
|
||||
if not current_price or current_price <= 0:
|
||||
continue
|
||||
analysis_price = float(row.get("price_at_analysis") or 0.0)
|
||||
if analysis_price <= 0:
|
||||
continue
|
||||
|
||||
return_pct = ((float(current_price) - analysis_price) / analysis_price) * 100.0
|
||||
decision = str(row.get("decision") or "HOLD")
|
||||
|
||||
was_correct = False
|
||||
if decision == "BUY" and return_pct > 2:
|
||||
was_correct = True
|
||||
elif decision == "SELL" and return_pct < -2:
|
||||
was_correct = True
|
||||
elif decision == "HOLD" and abs(return_pct) <= 5:
|
||||
was_correct = True
|
||||
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE qd_analysis_memory
|
||||
SET validated_at = NOW(),
|
||||
actual_return_pct = %s,
|
||||
was_correct = %s
|
||||
WHERE id = %s
|
||||
""",
|
||||
(return_pct, was_correct, int(row["id"])),
|
||||
)
|
||||
|
||||
stats["validated"] += 1
|
||||
if was_correct:
|
||||
stats["correct"] += 1
|
||||
else:
|
||||
stats["incorrect"] += 1
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to validate memory {row.get('id')}: {e}", exc_info=True)
|
||||
stats["errors"] += 1
|
||||
|
||||
db.commit()
|
||||
cur.close()
|
||||
except Exception as e:
|
||||
logger.error(f"validate_unvalidated_older_than failed: {e}", exc_info=True)
|
||||
|
||||
return stats
|
||||
|
||||
def get_performance_stats(self, market: str = None, symbol: str = None,
|
||||
days: int = 30) -> Dict[str, Any]:
|
||||
|
||||
@@ -9,6 +9,7 @@ Fast Analysis Service 3.0
|
||||
4. 单次LLM调用 - 强约束prompt，输出结构化分析
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from typing import Dict, Any, Optional, List
|
||||
from decimal import Decimal, ROUND_HALF_UP
|
||||
@@ -37,7 +38,17 @@ class FastAnalysisService:
|
||||
|
||||
# ==================== Data Collection Layer ====================
|
||||
|
||||
def _collect_market_data(self, market: str, symbol: str, timeframe: str = "1D") -> Dict[str, Any]:
|
||||
def _collect_market_data(
|
||||
self,
|
||||
market: str,
|
||||
symbol: str,
|
||||
timeframe: str = "1D",
|
||||
*,
|
||||
include_macro: bool = True,
|
||||
include_news: bool = True,
|
||||
include_polymarket: bool = True,
|
||||
timeout: int = 45,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
使用统一的数据采集器收集市场数据
|
||||
|
||||
@@ -52,10 +63,10 @@ class FastAnalysisService:
|
||||
market=market,
|
||||
symbol=symbol,
|
||||
timeframe=timeframe,
|
||||
include_macro=True,
|
||||
include_news=True,
|
||||
include_polymarket=True, # 包含预测市场数据
|
||||
timeout=45 # 增加超时时间，确保数据收集完成
|
||||
include_macro=include_macro,
|
||||
include_news=include_news,
|
||||
include_polymarket=include_polymarket, # 包含预测市场数据
|
||||
timeout=timeout, # 增加超时时间，确保数据收集完成
|
||||
)
|
||||
|
||||
def _calculate_indicators(self, kline_data: List[Dict]) -> Dict[str, Any]:
|
||||
@@ -694,9 +705,148 @@ IMPORTANT:
|
||||
}
|
||||
|
||||
try:
|
||||
# Phase 1: Data collection (parallel)
|
||||
# Phase 1: Data collection (multi-timeframe for consensus)
|
||||
logger.info(f"Fast analysis starting: {market}:{symbol}")
|
||||
data = self._collect_market_data(market, symbol, timeframe)
|
||||
|
||||
# Consensus timeframes:
|
||||
# - 默认：用用户传入的 timeframe 作为主周期，再加一个上层周期（1D/4H）提升稳定性
|
||||
# - 也允许通过 env 覆盖（逗号分隔），例如 AI_ANALYSIS_CONSENSUS_TIMEFRAMES=1D,4H
|
||||
env_tfs = os.getenv("AI_ANALYSIS_CONSENSUS_TIMEFRAMES", "").strip()
|
||||
if env_tfs:
|
||||
consensus_timeframes = [t.strip() for t in env_tfs.split(",") if t.strip()]
|
||||
else:
|
||||
# Heuristic defaults
|
||||
tf0 = (timeframe or "").strip().upper()
|
||||
# Primary first
|
||||
consensus_timeframes = [tf0] if tf0 else [timeframe]
|
||||
# Add 4H/1D depending on primary
|
||||
if tf0 in ("1H", "1HOUR", "60M"):
|
||||
consensus_timeframes += ["4H", "1D"]
|
||||
elif tf0 in ("4H",):
|
||||
consensus_timeframes += ["1D"]
|
||||
elif tf0 in ("1D", "1DAY", "D"):
|
||||
consensus_timeframes += ["4H"]
|
||||
else:
|
||||
# Generic fallback
|
||||
consensus_timeframes += ["1D", "4H"]
|
||||
# Dedup keep order
|
||||
seen = set()
|
||||
consensus_timeframes = [x for x in consensus_timeframes if not (x in seen or seen.add(x))]
|
||||
|
||||
primary_tf = (timeframe or "").strip().upper() or "1D"
|
||||
# Always include the primary timeframe in consensus,
|
||||
# even when env overrides timeframes.
|
||||
if primary_tf and primary_tf not in consensus_timeframes:
|
||||
consensus_timeframes = [primary_tf] + list(consensus_timeframes)
|
||||
# De-dup keep order
|
||||
seen = set()
|
||||
consensus_timeframes = [x for x in consensus_timeframes if not (x in seen or seen.add(x))]
|
||||
# Collect primary data including macro/news/polymarket for prompt quality
|
||||
primary_data = self._collect_market_data(
|
||||
market,
|
||||
symbol,
|
||||
primary_tf,
|
||||
include_macro=True,
|
||||
include_news=True,
|
||||
include_polymarket=True,
|
||||
)
|
||||
|
||||
# Collect extra timeframes for objective consensus (technical-only for cost)
|
||||
objective_by_tf: Dict[str, Dict[str, Any]] = {}
|
||||
decision_votes: Dict[str, int] = {"BUY": 0, "SELL": 0, "HOLD": 0}
|
||||
weighted_score_sum = 0.0
|
||||
weighted_score_w_sum = 0.0
|
||||
|
||||
def _extract_current_price(d: Dict[str, Any]) -> Optional[float]:
|
||||
if d.get("price") and d["price"].get("price"):
|
||||
try:
|
||||
return float(d["price"]["price"])
|
||||
except Exception:
|
||||
return None
|
||||
ind = d.get("indicators") or {}
|
||||
cp = ind.get("current_price")
|
||||
try:
|
||||
if cp:
|
||||
return float(cp)
|
||||
except Exception:
|
||||
pass
|
||||
# fallback to kline close
|
||||
kl = d.get("kline") or []
|
||||
if kl:
|
||||
try:
|
||||
return float(kl[-1].get("close") or 0)
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
logger.info(f"Consensus timeframes: {consensus_timeframes}")
|
||||
for tf in consensus_timeframes:
|
||||
tf_norm = (tf or "").strip().upper()
|
||||
if not tf_norm:
|
||||
continue
|
||||
|
||||
if tf_norm == primary_tf:
|
||||
d_tf = primary_data
|
||||
else:
|
||||
d_tf = self._collect_market_data(
|
||||
market,
|
||||
symbol,
|
||||
tf_norm,
|
||||
include_macro=False,
|
||||
include_news=False,
|
||||
include_polymarket=False,
|
||||
timeout=25,
|
||||
)
|
||||
|
||||
current_price_tf = _extract_current_price(d_tf) or 0.0
|
||||
objective = self._calculate_objective_score(d_tf, current_price_tf)
|
||||
overall_score = float(objective.get("overall_score", 0.0) or 0.0)
|
||||
decision = self._score_to_decision(overall_score, market=market)
|
||||
abs_score = abs(overall_score)
|
||||
|
||||
objective_by_tf[tf_norm] = {
|
||||
"objective_score": objective,
|
||||
"overall_score": overall_score,
|
||||
"decision": decision,
|
||||
"abs_score": abs_score,
|
||||
}
|
||||
decision_votes[decision] = decision_votes.get(decision, 0) + 1
|
||||
|
||||
# Weight by strength so strong cycles dominate
|
||||
w = 1.0 + min(1.5, abs_score / 100.0)
|
||||
weighted_score_sum += overall_score * w
|
||||
weighted_score_w_sum += w
|
||||
|
||||
consensus_score = weighted_score_sum / weighted_score_w_sum if weighted_score_w_sum > 0 else 0.0
|
||||
consensus_decision = self._score_to_decision(consensus_score, market=market)
|
||||
consensus_abs = abs(consensus_score)
|
||||
|
||||
# Agreement factor: how many timeframes support the consensus decision
|
||||
tf_count = max(1, len(objective_by_tf))
|
||||
agreement_cnt = sum(1 for x in objective_by_tf.values() if str(x.get("decision") or "").upper() == consensus_decision)
|
||||
agreement_ratio = agreement_cnt / tf_count
|
||||
|
||||
# Data quality degradation: derive from primary_data meta
|
||||
meta = primary_data.get("_meta") or {}
|
||||
failed_items = set(meta.get("failed_items") or [])
|
||||
quality_multiplier = 1.0
|
||||
if "macro" in failed_items:
|
||||
quality_multiplier *= 0.85
|
||||
if "news" in failed_items:
|
||||
quality_multiplier *= 0.8
|
||||
if "polymarket" in failed_items:
|
||||
quality_multiplier *= 0.9
|
||||
# If indicators missing key sections, reduce confidence more
|
||||
ind = primary_data.get("indicators") or {}
|
||||
if not ind or not ind.get("rsi") or not ind.get("moving_averages"):
|
||||
quality_multiplier *= 0.65
|
||||
|
||||
logger.info(
|
||||
f"Consensus decision={consensus_decision}, score={consensus_score:.2f}, "
|
||||
f"agreement_ratio={agreement_ratio:.2f}, quality_multiplier={quality_multiplier:.2f}"
|
||||
)
|
||||
|
||||
data = primary_data # keep original variable usage for prompt/LLM input
|
||||
|
||||
# Validate we have essential data - with fallback to indicators
|
||||
current_price = None
|
||||
@@ -771,35 +921,69 @@ IMPORTANT:
|
||||
llm_time = int((time.time() - llm_start) * 1000)
|
||||
logger.info(f"LLM call completed in {llm_time}ms")
|
||||
|
||||
# Phase 4: Calculate objective score and determine decision based on score
|
||||
# Phase 4: Objective score (primary tf) + consensus calibration
|
||||
objective_score = self._calculate_objective_score(data, current_price)
|
||||
logger.info(f"Objective score calculated: {objective_score['overall_score']:.1f} (Technical: {objective_score['technical_score']:.1f}, Fundamental: {objective_score['fundamental_score']:.1f}, Sentiment: {objective_score['sentiment_score']:.1f}, Macro: {objective_score['macro_score']:.1f})")
|
||||
|
||||
# Determine decision based on objective score thresholds
|
||||
score_based_decision = self._score_to_decision(objective_score['overall_score'])
|
||||
logger.info(f"Score-based decision: {score_based_decision} (score: {objective_score['overall_score']:.1f})")
|
||||
|
||||
# Override LLM decision with score-based decision if they differ significantly
|
||||
llm_decision = analysis.get("decision", "HOLD")
|
||||
if llm_decision != score_based_decision:
|
||||
score_abs = abs(objective_score['overall_score'])
|
||||
# 降低阈值，因为现在HOLD区间更小了（±20），±15以上的评分就应该覆盖
|
||||
if score_abs >= 15: # 如果评分达到±15以上，就覆盖LLM决策（因为阈值是±20）
|
||||
logger.warning(f"LLM decision '{llm_decision}' conflicts with score-based decision '{score_based_decision}' (score: {objective_score['overall_score']:.1f}). Overriding to score-based decision.")
|
||||
analysis["decision"] = score_based_decision
|
||||
# Adjust confidence based on score strength
|
||||
# 评分越高，置信度越高（最高95，最低60）
|
||||
analysis["confidence"] = min(95, max(60, int(50 + score_abs * 0.45)))
|
||||
# Update summary to mention score-based decision
|
||||
original_summary = analysis.get("summary", "")
|
||||
score_level = "强烈" if score_abs >= 70 else "明显" if score_abs >= 40 else "轻微"
|
||||
analysis["summary"] = f"{original_summary} [基于客观评分系统：综合评分{objective_score['overall_score']:.1f}分（{score_level}{'利多' if objective_score['overall_score'] > 0 else '利空'}），建议{score_based_decision}]"
|
||||
else:
|
||||
logger.info(f"LLM decision '{llm_decision}' differs from score-based '{score_based_decision}' but score is close to neutral ({objective_score['overall_score']:.1f}), keeping LLM decision")
|
||||
|
||||
# Add objective scores to analysis
|
||||
logger.info(
|
||||
f"Primary objective score: {objective_score['overall_score']:.1f} "
|
||||
f"(Technical: {objective_score['technical_score']:.1f}, Fundamental: {objective_score['fundamental_score']:.1f}, "
|
||||
f"Sentiment: {objective_score['sentiment_score']:.1f}, Macro: {objective_score['macro_score']:.1f})"
|
||||
)
|
||||
|
||||
score_based_decision = self._score_to_decision(objective_score["overall_score"], market=market)
|
||||
llm_decision = str(analysis.get("decision", "HOLD") or "HOLD").upper()
|
||||
|
||||
# Consensus confidence:
|
||||
consensus_conf = int(max(40, min(98, 50 + consensus_abs * 0.35)))
|
||||
# Agreement boosts, disagreement reduces
|
||||
consensus_conf = int(max(35, min(98, consensus_conf * (0.85 + 0.3 * agreement_ratio))))
|
||||
consensus_conf = int(max(0, min(100, consensus_conf * quality_multiplier)))
|
||||
|
||||
# Decide whether to enforce consensus over LLM / primary-score decision
|
||||
cfg = self._get_ai_calibration(market=market)
|
||||
min_abs_override = float(cfg.get("min_consensus_abs_override") or 15.0)
|
||||
quality_hold_thr = float(cfg.get("quality_hold_threshold") or 0.7)
|
||||
|
||||
if consensus_abs >= min_abs_override:
|
||||
final_decision = consensus_decision
|
||||
if llm_decision != final_decision:
|
||||
logger.warning(
|
||||
f"Override: llm_decision={llm_decision}, consensus_decision={final_decision}, "
|
||||
f"consensus_score={consensus_score:.1f}, consensus_abs={consensus_abs:.1f}"
|
||||
)
|
||||
analysis["decision"] = final_decision
|
||||
analysis["confidence"] = consensus_conf
|
||||
original_summary = analysis.get("summary", "")
|
||||
level = "强烈" if consensus_abs >= 70 else "明显" if consensus_abs >= 40 else "轻微"
|
||||
analysis["summary"] = (
|
||||
f"{original_summary} [多周期客观共识：综合评分{consensus_score:.1f}分（"
|
||||
f"{level}{'利多' if consensus_score > 0 else '利空'}），建议{final_decision}]"
|
||||
)
|
||||
else:
|
||||
# Near-neutral: keep LLM but shrink confidence by quality and enforce HOLD if quality is poor
|
||||
analysis["confidence"] = int(max(0, min(100, int(analysis.get("confidence", 50) or 50) * quality_multiplier)))
|
||||
|
||||
if quality_multiplier < quality_hold_thr:
|
||||
analysis["decision"] = "HOLD"
|
||||
analysis["confidence"] = min(int(analysis.get("confidence", 50) or 50), 55)
|
||||
|
||||
# Add objective scores and consensus to analysis
|
||||
analysis["objective_score"] = objective_score
|
||||
analysis["score_based_decision"] = score_based_decision
|
||||
analysis["objective_scores_by_timeframe"] = {
|
||||
k: {
|
||||
"overall_score": v.get("overall_score"),
|
||||
"decision": v.get("decision"),
|
||||
"abs_score": v.get("abs_score"),
|
||||
}
|
||||
for k, v in objective_by_tf.items()
|
||||
}
|
||||
analysis["consensus"] = {
|
||||
"consensus_score": consensus_score,
|
||||
"consensus_decision": consensus_decision,
|
||||
"consensus_abs": consensus_abs,
|
||||
"agreement_ratio": agreement_ratio,
|
||||
"quality_multiplier": quality_multiplier,
|
||||
}
|
||||
|
||||
# Phase 5: Validate and constrain output (pass indicators for decision validation)
|
||||
# Check for major news or macro events that could override technical indicators
|
||||
@@ -815,6 +999,21 @@ IMPORTANT:
|
||||
has_major_news=has_major_news,
|
||||
has_macro_event=has_macro_event
|
||||
)
|
||||
|
||||
# Post-validate: adjust position sizing based on quality + agreement
|
||||
try:
|
||||
ps = analysis.get("position_size_pct", 10)
|
||||
ps = int(float(ps or 10))
|
||||
# Lower position size if data is incomplete or multi-timeframe disagreement exists
|
||||
# agreement_ratio in [0..1]
|
||||
agreement_scale = 0.6 + 0.4 * float(agreement_ratio)
|
||||
ps_scaled = ps * float(quality_multiplier) * agreement_scale
|
||||
if str(analysis.get("decision") or "").upper() == "HOLD":
|
||||
ps_scaled *= 0.25
|
||||
analysis["position_size_pct"] = max(1, min(100, int(round(ps_scaled))))
|
||||
except Exception:
|
||||
# Keep model-provided position_size_pct
|
||||
pass
|
||||
|
||||
# Build final result
|
||||
total_time = int((time.time() - start_time) * 1000)
|
||||
@@ -860,6 +1059,7 @@ IMPORTANT:
|
||||
"resistance": data["indicators"].get("levels", {}).get("resistance"),
|
||||
},
|
||||
"indicators": data.get("indicators", {}),
|
||||
"consensus": analysis.get("consensus", {}),
|
||||
"analysis_time_ms": total_time,
|
||||
"llm_time_ms": llm_time,
|
||||
"data_collection_time_ms": data.get("collection_time_ms", 0),
|
||||
@@ -1207,14 +1407,39 @@ IMPORTANT:
|
||||
macro_score = self._calculate_macro_score(macro, data.get("market", ""))
|
||||
|
||||
# 5. 综合评分（加权平均）
|
||||
# 优化权重：技术35%，基本面20%，情绪25%（包含地缘政治），宏观20%（提高宏观权重）
|
||||
# 提高情绪和宏观权重，因为地缘政治和宏观经济因素对市场影响更大
|
||||
overall_score = (
|
||||
technical_score * 0.35 +
|
||||
fundamental_score * 0.20 +
|
||||
sentiment_score * 0.25 + # 提高情绪权重，包含地缘政治事件
|
||||
macro_score * 0.20 # 提高宏观权重
|
||||
)
|
||||
# 优化权重：默认技术35%，基本面20%，情绪25%（包含地缘政治），宏观20%（提高宏观权重）
|
||||
# 但要做“可用信息重加权”：当某些模块缺失（如新闻/宏观没取到），不要用0分去稀释整体强度，
|
||||
# 而是重新归一化权重，让技术信号在缺失时仍可发挥主导作用。
|
||||
market_type = str(data.get("market") or "")
|
||||
fundamental_present = (market_type == "USStock") and bool(fundamental)
|
||||
sentiment_present = bool(news)
|
||||
macro_present = bool(macro)
|
||||
# indicators 一旦成功计算通常就存在，但这里也做一次保护
|
||||
technical_present = bool(indicators)
|
||||
|
||||
weights = {
|
||||
"technical": 0.35,
|
||||
"fundamental": 0.20,
|
||||
"sentiment": 0.25,
|
||||
"macro": 0.20,
|
||||
}
|
||||
present_flags = {
|
||||
"technical": technical_present,
|
||||
"fundamental": fundamental_present,
|
||||
"sentiment": sentiment_present,
|
||||
"macro": macro_present,
|
||||
}
|
||||
|
||||
total_w = sum(w for k, w in weights.items() if present_flags.get(k))
|
||||
if total_w <= 0:
|
||||
overall_score = technical_score
|
||||
else:
|
||||
overall_score = (
|
||||
(technical_score * weights["technical"] if present_flags.get("technical") else 0.0)
|
||||
+ (fundamental_score * weights["fundamental"] if present_flags.get("fundamental") else 0.0)
|
||||
+ (sentiment_score * weights["sentiment"] if present_flags.get("sentiment") else 0.0)
|
||||
+ (macro_score * weights["macro"] if present_flags.get("macro") else 0.0)
|
||||
) / total_w
|
||||
|
||||
return {
|
||||
"technical_score": technical_score,
|
||||
@@ -1223,6 +1448,34 @@ IMPORTANT:
|
||||
"macro_score": macro_score,
|
||||
"overall_score": overall_score
|
||||
}
|
||||
|
||||
def _get_ai_calibration(self, market: str = "Crypto") -> Dict[str, Any]:
|
||||
"""
|
||||
Load latest offline calibration thresholds for the given market.
|
||||
Cached briefly to avoid DB load on every request.
|
||||
"""
|
||||
# Simple per-process cache
|
||||
now = time.time()
|
||||
if not hasattr(self, "_calibration_cache"):
|
||||
self._calibration_cache = {}
|
||||
self._calibration_cache_ts = {}
|
||||
ttl = int(os.getenv("AI_CALIBRATION_CACHE_TTL_SEC", "300"))
|
||||
key = (market or "").strip() or "Crypto"
|
||||
ts = self._calibration_cache_ts.get(key) or 0.0
|
||||
if ts and (now - float(ts)) < ttl:
|
||||
return self._calibration_cache.get(key) or {}
|
||||
|
||||
try:
|
||||
from app.services.ai_calibration import AICalibrationService
|
||||
svc = AICalibrationService()
|
||||
cfg = svc.get_latest(key)
|
||||
except Exception as e:
|
||||
logger.warning(f"_get_ai_calibration failed (fallback): {e}", exc_info=True)
|
||||
cfg = {}
|
||||
|
||||
self._calibration_cache[key] = cfg
|
||||
self._calibration_cache_ts[key] = now
|
||||
return cfg
|
||||
|
||||
def _calculate_technical_score(self, indicators: Dict, price_data: Dict) -> float:
|
||||
"""计算技术指标评分 (-100 to +100)"""
|
||||
@@ -1288,6 +1541,94 @@ IMPORTANT:
|
||||
change_score = change_24h * 2 # 线性映射
|
||||
score += change_score * 0.20
|
||||
weight_sum += 0.20
|
||||
|
||||
# ========== 额外技术特征（轻量增强，不改变主体结构） ==========
|
||||
# 这些特征来自 MarketDataCollector._calculate_indicators 的输出：
|
||||
# - price_position: 过去20根K线区间位置 0~100
|
||||
# - volume_ratio: 最新成交量 / 20期均量
|
||||
# - bollinger: BB_upper/BB_lower/BB_width
|
||||
# - volatility: atr, pct
|
||||
extra_score = 0.0
|
||||
extra_weight = 0.0
|
||||
|
||||
# 1) 区间位置：接近区间顶部更偏利空，接近区间底部更偏利多
|
||||
try:
|
||||
pp = float(indicators.get("price_position", 50.0))
|
||||
# 0~100 -> -15~+15 (线性映射，中心50为0)
|
||||
pp_score = (50.0 - pp) * 0.3
|
||||
# 在极端区域增强信号
|
||||
if pp >= 85:
|
||||
pp_score -= 5
|
||||
elif pp <= 15:
|
||||
pp_score += 5
|
||||
extra_score += pp_score
|
||||
extra_weight += 0.20
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 2) 布林带触及：突破上轨偏利空，跌破下轨偏利多
|
||||
try:
|
||||
cur_px = float(indicators.get("current_price") or price_data.get("price") or 0.0)
|
||||
bb = indicators.get("bollinger") or {}
|
||||
bb_u = float(bb.get("BB_upper") or 0.0)
|
||||
bb_l = float(bb.get("BB_lower") or 0.0)
|
||||
if cur_px > 0 and bb_u > 0 and bb_l > 0 and bb_u > bb_l:
|
||||
if cur_px >= bb_u:
|
||||
extra_score += -12
|
||||
extra_weight += 0.20
|
||||
elif cur_px <= bb_l:
|
||||
extra_score += +12
|
||||
extra_weight += 0.20
|
||||
else:
|
||||
# Within bands: small contribution by relative position
|
||||
rel = (cur_px - bb_l) / (bb_u - bb_l) # 0..1
|
||||
extra_score += (0.5 - float(rel)) * 10
|
||||
extra_weight += 0.10
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 3) 成交量放大：在趋势方向上加分，逆趋势减分（弱信号）
|
||||
try:
|
||||
vr = float(indicators.get("volume_ratio") or 1.0)
|
||||
trend = str(indicators.get("trend") or indicators.get("moving_averages", {}).get("trend") or "").lower()
|
||||
if vr >= 1.8:
|
||||
if "uptrend" in trend:
|
||||
extra_score += +8
|
||||
extra_weight += 0.15
|
||||
elif "downtrend" in trend:
|
||||
extra_score += -8
|
||||
extra_weight += 0.15
|
||||
else:
|
||||
# 放量但无趋势：更偏不确定，略微降低（当作偏利空风险）
|
||||
extra_score += -3
|
||||
extra_weight += 0.10
|
||||
elif vr <= 0.6:
|
||||
# 缩量：趋势信号可信度下降（轻微回归到0）
|
||||
extra_score += 0
|
||||
extra_weight += 0.05
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 4) 高波动：减少强方向自信（用“缩放”形式实现，避免硬反转）
|
||||
try:
|
||||
vol = indicators.get("volatility") or {}
|
||||
vol_pct = float(vol.get("pct") or 0.0)
|
||||
if vol_pct >= 6.0:
|
||||
# 极高波动：把额外分数打折，并轻微把总体拉回0
|
||||
extra_score *= 0.6
|
||||
score *= 0.92
|
||||
elif vol_pct >= 3.5:
|
||||
extra_score *= 0.8
|
||||
score *= 0.96
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Combine extra into main score (treat as another component)
|
||||
if extra_weight > 0:
|
||||
# Normalize extra to roughly -100..+100 scale
|
||||
extra_norm = max(-100.0, min(100.0, float(extra_score)))
|
||||
score += extra_norm * 0.15
|
||||
weight_sum += 0.15
|
||||
|
||||
# 归一化到-100到+100
|
||||
if weight_sum > 0:
|
||||
@@ -1536,17 +1877,38 @@ IMPORTANT:
|
||||
tnx_score = 0
|
||||
score += tnx_score
|
||||
factors += 1
|
||||
|
||||
# 恐惧贪婪指数（更适合 Crypto）：极端贪婪偏利空，极端恐惧偏利多（弱信号）
|
||||
try:
|
||||
fg = macro.get("FEAR_GREED", {}) or {}
|
||||
fg_value = float(fg.get("price") or 0.0)
|
||||
if fg_value > 0 and market in ["Crypto"]:
|
||||
if fg_value >= 80:
|
||||
score += -15
|
||||
factors += 1
|
||||
elif fg_value >= 65:
|
||||
score += -8
|
||||
factors += 1
|
||||
elif fg_value <= 20:
|
||||
score += +10
|
||||
factors += 1
|
||||
elif fg_value <= 35:
|
||||
score += +5
|
||||
factors += 1
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 归一化（考虑权重）
|
||||
if factors > 0:
|
||||
# 最大可能分数：VIX(-50~+20), DXY(-30~+30), TNX(-30~+30) = 约-110到+80
|
||||
# 归一化到-100到+100
|
||||
max_possible = 110 # 最大绝对值
|
||||
# 加上 Fear&Greed 的幅度（约 15），给点 buffer
|
||||
max_possible = 125 # 最大绝对值
|
||||
score = score / max_possible * 100
|
||||
|
||||
return max(-100, min(100, score))
|
||||
|
||||
def _score_to_decision(self, score: float) -> str:
|
||||
def _score_to_decision(self, score: float, *, market: str = "Crypto") -> str:
|
||||
"""
|
||||
根据客观评分转换为决策
|
||||
|
||||
@@ -1566,10 +1928,13 @@ IMPORTANT:
|
||||
- -70 < score <= -40: 明显SELL
|
||||
- score <= -70: 强烈SELL
|
||||
"""
|
||||
# 使用±20作为主要阈值，大幅缩小HOLD区间
|
||||
if score >= 20:
|
||||
cfg = self._get_ai_calibration(market=market)
|
||||
buy_thr = float(cfg.get("buy_threshold") or 20.0)
|
||||
sell_thr = float(cfg.get("sell_threshold") or -20.0)
|
||||
|
||||
if score >= buy_thr:
|
||||
return "BUY"
|
||||
elif score <= -20:
|
||||
elif score <= sell_thr:
|
||||
return "SELL"
|
||||
else:
|
||||
return "HOLD"
|
||||
|
||||
Reference in New Issue
Block a user