BluefxPraise/The_Oracle
0
1"""Brain: orchestrates scans, signals, learning phases, broadcasts."""2 3import asyncio4import json5import logging6import threading7from datetime import datetime, timezone, timedelta8from typing import Any, Dict, List, Optional9 10import requests11 12import memory13from config import (14 PAIRS, MAX_DAILY_SIGNALS_PER_PAIR, OBSERVATION_DAYS, ADMIN_CHAT_ID,15 TELEGRAM_BOT_TOKEN, INITIAL_CONFIDENCE_THRESHOLD,16)17from core import ensemble, regime_detector, ml_classifier18from core.risk_engine import compute_sl_tp, atr_value, is_halted, current_risk_pct19from interface.formatting import fmt_signal20from modules import news_shield, trade_reviewer, auto_reports21from utils.data_fetcher import get_candles, get_latest_price22from utils.scheduler import in_quiet_hours23from exports.csv_exporter import export_signals_csv24from exports.json_exporter import export_signals_json25from exports.pdf_exporter import export_signals_pdf26from backtest.parameter_tuner import sweep as param_sweep27from backtest.walkforward import walk_forward28 29log = logging.getLogger("oracle.brain")30 31 32def _tg_send(chat_id: str, text: str, parse_mode: Optional[str] = "MarkdownV2",33 disable_notification: bool = False) -> bool:34 try:35 r = requests.post(36 f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",37 json={38 "chat_id": chat_id,39 "text": text,40 "parse_mode": parse_mode,41 "disable_notification": disable_notification,42 "disable_web_page_preview": True,43 },44 timeout=15,45 )46 if r.status_code != 200:47 log.warning("tg send %s -> %s %s", chat_id, r.status_code, r.text[:200])48 return False49 return True50 except Exception as e:51 log.warning("tg send err: %s", e)52 return False53 54 55def _tg_send_doc(chat_id: str, file_path: str, caption: str = "") -> bool:56 try:57 with open(file_path, "rb") as fh:58 r = requests.post(59 f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendDocument",60 data={"chat_id": chat_id, "caption": caption},61 files={"document": fh},62 timeout=60,63 )64 return r.status_code == 20065 except Exception as e:66 log.warning("tg send doc err: %s", e)67 return False68 69 70class Brain:71 def __init__(self):72 self._lock = threading.RLock()73 self._first_run_init()74 75 # ---------- init ----------76 77 def _first_run_init(self):78 if memory.get_state("phase") is None:79 memory.set_state("phase", "OBSERVATION")80 memory.set_state("phase_started", datetime.now(timezone.utc).isoformat())81 memory.set_state("confidence_threshold", INITIAL_CONFIDENCE_THRESHOLD)82 log.info("first-run: starting in OBSERVATION phase")83 84 # ---------- public phase API ----------85 86 def phase(self) -> str:87 return memory.get_state("phase", "OBSERVATION")88 89 def set_phase(self, p: str):90 memory.set_state("phase", p)91 memory.set_state("phase_started", datetime.now(timezone.utc).isoformat())92 log.info("phase -> %s", p)93 94 def go_live(self):95 if self.phase() != "LIVE":96 self.set_phase("LIVE")97 self._notify_admin("🚀 *Oracle is now LIVE\\.* Signals will broadcast to active targets\\.")98 99 def promote(self) -> dict:100 """Manually advance one phase: OBSERVATION → AUTO_CALIBRATION → LIVE."""101 cur = self.phase()102 if cur == "OBSERVATION":103 self.set_phase("AUTO_CALIBRATION")104 self._notify_admin("🔧 Manual promote\\. Running AUTO\\_CALIBRATION\\.\\.\\.")105 try:106 self.calibrate()107 self.set_phase("LIVE")108 self._notify_admin("✅ Calibration done\\. Oracle is now *LIVE*\\.")109 return {"from": "OBSERVATION", "to": "LIVE", "calibrated": True}110 except Exception as e:111 log.exception("promote calibrate failed")112 return {"from": "OBSERVATION", "to": "AUTO_CALIBRATION", "error": str(e)}113 if cur == "AUTO_CALIBRATION":114 self.set_phase("LIVE")115 self._notify_admin("🚀 *Oracle is now LIVE\\.*")116 return {"from": "AUTO_CALIBRATION", "to": "LIVE"}117 return {"from": cur, "to": cur, "noop": True}118 119 def check_phase_transition(self):120 if self.phase() != "OBSERVATION":121 return122 started_str = memory.get_state("phase_started")123 if not started_str:124 return125 try:126 started = datetime.fromisoformat(started_str)127 except Exception:128 return129 days = (datetime.now(timezone.utc) - started).days130 self._notify_admin(131 f"📅 OBSERVATION day {days}/{OBSERVATION_DAYS} — silent learning in progress\\."132 )133 if days >= OBSERVATION_DAYS:134 self.set_phase("AUTO_CALIBRATION")135 self._notify_admin("🔧 Observation complete\\. Running AUTO\\_CALIBRATION\\.\\.\\.")136 try:137 self.calibrate()138 self.set_phase("LIVE")139 self._notify_admin("✅ Auto\\-calibration done\\. Oracle is now *LIVE*\\.")140 except Exception as e:141 log.exception("auto calibration failed")142 self._notify_admin(f"⚠️ Calibration error: {str(e)[:200]}")143 144 # ---------- scan + signal ----------145 146 def scan_all(self):147 try:148 if in_quiet_hours():149 log.info("quiet hours, skipping scan")150 return151 for pair in PAIRS:152 try:153 self.scan_pair(pair)154 except Exception as e:155 log.exception("scan_pair %s failed: %s", pair, e)156 except Exception:157 log.exception("scan_all crashed")158 159 def scan_pair(self, pair: str):160 if memory.signals_today(pair) >= MAX_DAILY_SIGNALS_PER_PAIR:161 return162 candles_15m = get_candles(pair, "15m", limit=250)163 if not candles_15m or len(candles_15m) < 60:164 return165 candles_1h = get_candles(pair, "1h", limit=200)166 candles_4h = get_candles(pair, "4h", limit=200)167 regime = regime_detector.classify(pair, candles_1h, candles_4h)168 if regime.get("regime_shift"):169 self._notify_admin(f"🌀 Regime shift on `{pair}`: {regime['regime_shift']}")170 171 sig = ensemble.evaluate(pair, candles_15m)172 if not sig:173 return174 sig["regime"] = regime["combined"]175 176 sig = news_shield.filter_signal(sig)177 if not sig:178 return179 if sig["confidence"] < ensemble.get_threshold():180 return181 182 entry = float(candles_15m[-1]["c"])183 atr = atr_value(candles_15m, 14)184 sl, tp, rr = compute_sl_tp(sig["direction"], entry, atr, pair)185 sig.update({"entry": entry, "sl": sl, "tp": tp, "rr": rr})186 187 observation = self.phase() != "LIVE"188 sid = memory.insert_signal(sig, observation=observation)189 sig["id"] = sid190 log.info("signal %d %s %s @ %.5f conf=%.1f obs=%s",191 sid, pair, sig["direction"], entry, sig["confidence"], observation)192 193 if not observation and not is_halted():194 self.broadcast_signal(sig)195 memory.mark_signal_broadcast(sid)196 elif observation:197 self._notify_admin(198 f"🔬 OBS signal logged: `{pair}` {sig['direction']} conf {sig['confidence']:.0f}% \\(not broadcast\\)"199 )200 201 # ---------- broadcasting ----------202 203 def broadcast_signal(self, sig: Dict[str, Any]):204 text = fmt_signal(sig)205 targets = memory.list_targets(only_signals=True)206 for t in targets:207 ok = _tg_send(t["chat_id"], text, parse_mode="MarkdownV2")208 if not ok:209 log.warning("broadcast failed to %s", t["chat_id"])210 _tg_send(ADMIN_CHAT_ID, text, parse_mode="MarkdownV2", disable_notification=True)211 212 # ---------- open trades ----------213 214 def check_open_trades(self):215 try:216 for row in memory.open_signals():217 pair = row["pair"]218 price = get_latest_price(pair)219 if price <= 0:220 continue221 direction = row["direction"]222 entry, sl, tp = float(row["entry"]), float(row["sl"]), float(row["tp"])223 outcome = None224 pnl = 0.0225 if direction == "BUY":226 if price <= sl:227 outcome = "LOSS"; pnl = (sl - entry) / entry * 100228 elif price >= tp:229 outcome = "WIN"; pnl = (tp - entry) / entry * 100230 else:231 if price >= sl:232 outcome = "LOSS"; pnl = (entry - sl) / entry * 100233 elif price <= tp:234 outcome = "WIN"; pnl = (entry - tp) / entry * 100235 if outcome:236 memory.close_signal(row["id"], outcome, pnl)237 sig_dict = {k: row[k] for k in row.keys()}238 try:239 review = trade_reviewer.review_trade(sig_dict, outcome, pnl)240 except Exception as e:241 log.warning("reviewer failed: %s", e)242 review = None243 self._notify_admin(244 f"{'🟢' if outcome == 'WIN' else '🔴'} Trade closed `{pair}` {direction}: "245 f"*{outcome}* PnL `{pnl:+.2f}%`"246 )247 if review and review.get("rule"):248 self._notify_admin(f"🧠 New rule: _{review['rule'][:300]}_")249 except Exception:250 log.exception("check_open_trades crashed")251 252 # ---------- news ----------253 254 def refresh_news(self):255 try:256 news_shield.refresh(PAIRS)257 except Exception:258 log.exception("refresh_news crashed")259 260 # ---------- reports ----------261 262 def send_daily_report(self):263 try:264 text = auto_reports.daily_report()265 self._notify_admin(text, parse_mode=None)266 for t in memory.list_targets(only_reports=True):267 _tg_send(t["chat_id"], text, parse_mode=None)268 except Exception:269 log.exception("daily report failed")270 271 # ---------- calibration ----------272 273 def calibrate(self) -> Dict[str, Any]:274 results = {}275 # 1. recompute weights via Kelly per model from recent closed signals276 weights = self._recompute_weights_kelly()277 results["weights"] = weights278 279 # 2. sweep thresholds 60..90280 thr_metrics = self._sweep_threshold()281 results["threshold"] = thr_metrics282 283 # 3. sweep EMA combos + RSI (parameter_tuner)284 sweep_results = {}285 for p in PAIRS:286 candles = get_candles(p, "15m", limit=500)287 if len(candles) >= 80:288 sweep_results[p] = param_sweep(p, candles)289 results["param_sweep"] = {k: v["best"]["params"] for k, v in sweep_results.items() if v["best"]["params"]}290 291 # 4. retrain ML292 ml_classifier.retrain_all(PAIRS, lambda p, tf, n: get_candles(p, tf, n))293 results["ml_retrained"] = True294 295 memory.log_calibration("auto_calibration", {"phase": self.phase()}, results, applied=True)296 return results297 298 def _recompute_weights_kelly(self) -> Dict[str, float]:299 # Per-model winrate & avg-rr from closed signals' models payload300 conn = memory.get_conn()301 cur = conn.execute(302 "SELECT models_json, outcome FROM signals WHERE status='CLOSED' ORDER BY id DESC LIMIT 500"303 )304 rows = cur.fetchall()305 per: Dict[str, Dict[str, int]] = {"trend": {"w": 0, "l": 0}, "mr": {"w": 0, "l": 0}, "ml": {"w": 0, "l": 0}}306 for r in rows:307 try:308 models = json.loads(r["models_json"] or "{}")309 except Exception:310 continue311 for m_name in models.keys():312 if m_name not in per:313 per[m_name] = {"w": 0, "l": 0}314 if r["outcome"] == "WIN":315 per[m_name]["w"] += 1316 elif r["outcome"] == "LOSS":317 per[m_name]["l"] += 1318 weights = {}319 from config import ENSEMBLE_WEIGHTS320 for m, d in per.items():321 total = d["w"] + d["l"]322 if total < 5:323 weights[m] = float(ENSEMBLE_WEIGHTS.get(m, 1.0))324 else:325 p = d["w"] / total326 # simple Kelly with assumed b=1.5 (RR target)327 b = 1.5328 kelly = max(0.0, (p * (b + 1) - 1) / b)329 base = float(ENSEMBLE_WEIGHTS.get(m, 1.0))330 weights[m] = float(max(0.25, min(3.0, base * (0.5 + 2 * kelly))))331 wr = (d["w"] / total * 100) if total > 0 else 0.0332 memory.upsert_model_perf(m, None, d["w"], d["l"], wr, 1.5, weights[m])333 return weights334 335 def _sweep_threshold(self) -> Dict[str, Any]:336 conn = memory.get_conn()337 cur = conn.execute(338 "SELECT confidence, outcome FROM signals WHERE status='CLOSED' ORDER BY id DESC LIMIT 500"339 )340 rows = cur.fetchall()341 if len(rows) < 20:342 return {"chosen": ensemble.get_threshold(), "reason": "insufficient data"}343 best = (-1e9, ensemble.get_threshold())344 for thr in range(60, 95, 5):345 wins = sum(1 for r in rows if r["confidence"] >= thr and r["outcome"] == "WIN")346 losses = sum(1 for r in rows if r["confidence"] >= thr and r["outcome"] == "LOSS")347 n = wins + losses348 if n < 5:349 continue350 score = wins * 1.5 - losses351 if score > best[0]:352 best = (score, thr)353 memory.set_state("confidence_threshold", float(best[1]))354 return {"chosen": best[1], "score": best[0]}355 356 # ---------- weekly + monthly ----------357 358 def weekly_routine(self):359 try:360 csv_path = export_signals_csv()361 json_path = export_signals_json()362 pdf_path = export_signals_pdf()363 caption = f"Weekly Oracle exports — {datetime.now(timezone.utc).strftime('%Y-%m-%d')}"364 for t in [{"chat_id": ADMIN_CHAT_ID}] + list(memory.list_targets(only_reports=True)):365 _tg_send_doc(t["chat_id"], csv_path, caption=caption + " (CSV)")366 _tg_send_doc(t["chat_id"], json_path, caption=caption + " (JSON)")367 _tg_send_doc(t["chat_id"], pdf_path, caption=caption + " (PDF)")368 wk_text = auto_reports.weekly_report()369 self._notify_admin(wk_text, parse_mode=None)370 for t in memory.list_targets(only_reports=True):371 _tg_send(t["chat_id"], wk_text, parse_mode=None)372 # weekly micro-calibration373 self._notify_admin("🔧 Running weekly micro\\-calibration\\.\\.\\.")374 self.calibrate()375 self._notify_admin("✅ Weekly micro\\-calibration complete\\.")376 except Exception:377 log.exception("weekly_routine crashed")378 379 def monthly_deep_optimization(self):380 try:381 self._notify_admin("📅 Monthly deep optimization started\\.\\.\\.")382 results = {}383 for p in PAIRS:384 candles = get_candles(p, "15m", limit=1000)385 if len(candles) < 400:386 continue387 wf = walk_forward(p, candles, train_bars=300, test_bars=100,388 threshold=ensemble.get_threshold())389 results[p] = wf390 memory.log_calibration("monthly_deep", {"pairs": PAIRS}, results, applied=True)391 self._notify_admin("✅ Monthly deep optimization complete\\.")392 except Exception:393 log.exception("monthly_deep_optimization crashed")394 395 # ---------- helpers ----------396 397 def _notify_admin(self, text: str, parse_mode: Optional[str] = "MarkdownV2"):398 _tg_send(ADMIN_CHAT_ID, text, parse_mode=parse_mode, disable_notification=True)399