# ==============================================================================
# ๐ app.py (V61.4 - GEM-Architect: Full UI Integrity Restored)
# ==============================================================================
import os
import sys
import traceback
import asyncio
import gc
import time
import json
import logging
from datetime import datetime, timedelta
from contextlib import asynccontextmanager, redirect_stdout, redirect_stderr
from io import StringIO
from typing import List, Dict, Any, Optional
from fastapi import FastAPI, HTTPException, BackgroundTasks
import gradio as gr
import pandas as pd
import plotly.graph_objects as go
# ------------------------------------------------------------------------------
# Logging Setup
# ------------------------------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format="[%(levelname)s] %(message)s",
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("TitanCore")
# ------------------------------------------------------------------------------
# Imports
# ------------------------------------------------------------------------------
try:
from r2 import R2Service, INITIAL_CAPITAL
from ml_engine.data_manager import DataManager
from ml_engine.processor import MLProcessor, SystemLimits
from whale_monitor.core import EnhancedWhaleMonitor
from whale_monitor.rpc_manager import AdaptiveRpcManager
from sentiment_news import NewsFetcher
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
from learning_hub.adaptive_hub import AdaptiveHub
from trade_manager import TradeManager
from periodic_tuner import AutoTunerScheduler
try:
from backtest_engine import run_strategic_optimization_task
BACKTEST_AVAILABLE = True
except ImportError:
BACKTEST_AVAILABLE = False
except ImportError as e:
logger.critical(f"โ [FATAL ERROR] Failed to import core modules: {e}")
traceback.print_exc()
sys.exit(1)
# ------------------------------------------------------------------------------
# Global Context
# ------------------------------------------------------------------------------
r2: R2Service = None
data_manager: DataManager = None
ml_processor: MLProcessor = None
adaptive_hub: AdaptiveHub = None
trade_manager: TradeManager = None
whale_monitor: EnhancedWhaleMonitor = None
news_fetcher: NewsFetcher = None
senti_analyzer: SentimentIntensityAnalyzer = None
sys_state: 'SystemState' = None
scheduler: AutoTunerScheduler = None
# ------------------------------------------------------------------------------
# State Management
# ------------------------------------------------------------------------------
class SystemState:
def __init__(self):
self.ready = False
self.cycle_running = False
self.training_running = False
self.auto_pilot = True
self.last_cycle_time: datetime = None
self.last_cycle_error = None
self.app_start_time = datetime.now()
self.last_cycle_logs = "System Initializing..."
self.training_status_msg = "Adaptive Mode: Active"
self.scan_interval = 60
def set_ready(self):
self.ready = True
self.last_cycle_logs = "โ
System Ready. Cybernetic Loop ON."
logger.info("โ
System State set to READY.")
def set_cycle_start(self):
self.cycle_running = True
self.last_cycle_logs = "๐ [Cycle START] Scanning Markets..."
logger.info("๐ Cycle STARTED.")
def set_cycle_end(self, error=None, logs=None):
self.cycle_running = False
self.last_cycle_time = datetime.now()
self.last_cycle_error = str(error) if error else None
if logs:
self.last_cycle_logs = logs
elif error:
self.last_cycle_logs = f"โ [Cycle ERROR] {error}"
logger.error(f"Cycle Error: {error}")
else:
self.last_cycle_logs = f"โ
[Cycle END] Finished successfully."
logger.info("โ
Cycle ENDED.")
sys_state = SystemState()
# ------------------------------------------------------------------------------
# Utilities
# ------------------------------------------------------------------------------
def calculate_duration_str(timestamp_str):
if not timestamp_str: return "--:--:--"
try:
if isinstance(timestamp_str, str):
start_time = datetime.fromisoformat(timestamp_str)
else: start_time = timestamp_str
diff = datetime.now() - start_time
total_seconds = int(diff.total_seconds())
days = total_seconds // 86400
hours = (total_seconds % 86400) // 3600
minutes = (total_seconds % 3600) // 60
seconds = total_seconds % 60
if days > 0: return f"{days}d {hours:02}:{minutes:02}:{seconds:02}"
return f"{hours:02}:{minutes:02}:{seconds:02}"
except: return "--:--:--"
def format_pnl_split(profit, loss):
"""ุชูุณูู ุงูุฑุจุญูุฉ ูุฌุฒุฃูู (ุฑุจุญ / ุฎุณุงุฑุฉ) ุจุฃููุงู"""
return f"+${profit:,.0f} / -${abs(loss):,.0f}"
# ------------------------------------------------------------------------------
# Auto-Pilot Daemon
# ------------------------------------------------------------------------------
async def auto_pilot_loop():
logger.info("๐ค [Auto-Pilot] Daemon started.")
while True:
try:
await asyncio.sleep(5)
if not sys_state.ready: continue
if adaptive_hub and int(time.time()) % 60 == 0:
sys_state.training_status_msg = adaptive_hub.get_status()
if trade_manager and len(trade_manager.open_positions) > 0:
wd_status = await trade_manager.ensure_active_guardians()
if "No active" not in wd_status:
if not sys_state.cycle_running:
sys_state.last_cycle_logs = trade_manager.latest_guardian_log
continue
if sys_state.auto_pilot and not sys_state.cycle_running and not sys_state.training_running:
if sys_state.last_cycle_time:
elapsed = (datetime.now() - sys_state.last_cycle_time).total_seconds()
if elapsed < sys_state.scan_interval:
continue
logger.info("๐ค [Auto-Pilot] Triggering scan...")
asyncio.create_task(run_unified_cycle())
await asyncio.sleep(5)
except Exception as e:
logger.error(f"โ ๏ธ [Auto-Pilot Error] {e}")
await asyncio.sleep(30)
# ------------------------------------------------------------------------------
# Lifespan
# ------------------------------------------------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
global r2, data_manager, ml_processor, adaptive_hub, trade_manager, whale_monitor, news_fetcher, senti_analyzer, sys_state, scheduler
logger.info("\n๐ [System] Startup Sequence (Titan V61.4 - Restoration)...")
try:
r2 = R2Service()
data_manager = DataManager(contracts_db={}, whale_monitor=None, r2_service=r2)
await data_manager.initialize()
await data_manager.load_contracts_from_r2()
whale_monitor = EnhancedWhaleMonitor(contracts_db=data_manager.get_contracts_db(), r2_service=r2)
rpc_mgr = AdaptiveRpcManager(data_manager.http_client)
whale_monitor.set_rpc_manager(rpc_mgr)
news_fetcher = NewsFetcher()
senti_analyzer = SentimentIntensityAnalyzer()
data_manager.whale_monitor = whale_monitor
adaptive_hub = AdaptiveHub(r2_service=r2)
await adaptive_hub.initialize()
ml_processor = MLProcessor(data_manager=data_manager)
await ml_processor.initialize()
trade_manager = TradeManager(r2_service=r2, data_manager=data_manager, processor=ml_processor)
trade_manager.learning_hub = adaptive_hub
await trade_manager.initialize_sentry_exchanges()
await trade_manager.start_sentry_loops()
scheduler = AutoTunerScheduler(trade_manager)
asyncio.create_task(scheduler.start_loop())
logger.info("๐ฐ๏ธ [Scheduler] Auto-Tuner Background Task Started.")
sys_state.set_ready()
asyncio.create_task(auto_pilot_loop())
logger.info("โ
[System READY] All modules operational.")
yield
except Exception as e:
logger.critical(f"โ [FATAL STARTUP ERROR] {e}")
traceback.print_exc()
finally:
sys_state.ready = False
if trade_manager: await trade_manager.stop_sentry_loops()
if data_manager: await data_manager.close()
if whale_monitor and whale_monitor.rpc_manager: await whale_monitor.rpc_manager.close()
logger.info("โ
[System] Shutdown Complete.")
# ------------------------------------------------------------------------------
# Helper Tasks
# ------------------------------------------------------------------------------
async def _analyze_symbol_task(candidate_data: Dict[str, Any]) -> Dict[str, Any]:
try:
symbol = candidate_data['symbol']
required_tfs = ["5m", "15m", "1h", "4h"]
data_tasks = [data_manager.get_latest_ohlcv(symbol, tf, limit=300) for tf in required_tfs]
all_data = await asyncio.gather(*data_tasks)
ohlcv_data = {}
for tf, data in zip(required_tfs, all_data):
if data and len(data) > 0: ohlcv_data[tf] = data
if '1h' not in ohlcv_data or '5m' not in ohlcv_data:
return None
current_price = await data_manager.get_latest_price_async(symbol)
raw_data = {
'symbol': symbol,
'ohlcv': ohlcv_data,
'current_price': current_price,
'timestamp': time.time(),
'dynamic_limits': candidate_data.get('dynamic_limits', {}),
'asset_regime': candidate_data.get('asset_regime', 'UNKNOWN'),
'strategy_tag': candidate_data.get('strategy_tag', 'NONE'),
'strategy_type': candidate_data.get('strategy_type', 'NORMAL'),
'l1_score': candidate_data.get('l1_sort_score', 0)
}
res = await ml_processor.process_compound_signal(raw_data)
if not res: return None
res['strategy_type'] = candidate_data.get('strategy_type', 'NORMAL')
return res
except Exception: return None
# ------------------------------------------------------------------------------
# Unified Cycle
# ------------------------------------------------------------------------------
async def run_unified_cycle():
log_buffer = StringIO()
def log_and_print(message):
logger.info(message)
log_buffer.write(message + '\n')
if sys_state.cycle_running or sys_state.training_running: return
if not sys_state.ready: return
sys_state.set_cycle_start()
try:
await trade_manager.sync_internal_state_with_r2()
if len(trade_manager.open_positions) > 0:
log_and_print(f"โน๏ธ [Cycle] Active Positions: {len(trade_manager.open_positions)}")
for sym, tr in trade_manager.open_positions.items():
curr_p = await data_manager.get_latest_price_async(sym)
entry_p = float(tr.get('entry_price', 0))
pnl = ((curr_p - entry_p) / entry_p) * 100 if entry_p > 0 else 0
log_and_print(f" ๐ {sym}: {pnl:+.2f}%")
log_and_print(f" [1/5] ๐ L1 Screening (Bottom/Momentum)...")
candidates = await data_manager.layer1_rapid_screening(adaptive_hub_ref=adaptive_hub)
if not candidates:
log_and_print("โ ๏ธ No valid candidates found (Quality Filter).")
sys_state.set_cycle_end(logs=log_buffer.getvalue())
return
log_and_print(f" [2/5] ๐ง L2 Deep Analysis ({len(candidates)} items)...")
tasks = [_analyze_symbol_task(c) for c in candidates]
results = await asyncio.gather(*tasks)
valid_l2 = [res for res in results if res is not None]
semi_finalists = sorted(valid_l2, key=lambda x: x.get('enhanced_final_score', 0.0), reverse=True)[:10]
if not semi_finalists:
log_and_print("โ ๏ธ No valid L2 candidates.")
sys_state.set_cycle_end(logs=log_buffer.getvalue())
return
log_and_print(f" [3/5] ๐ก L3 Deep Dive (Whales & News) for TOP {len(semi_finalists)}...")
final_candidates = []
for sig in semi_finalists:
symbol = sig['symbol']
l2_score = sig.get('enhanced_final_score', 0.0)
whale_points = 0.0
try:
if whale_monitor:
w_data = await whale_monitor.get_symbol_whale_activity(symbol, known_price=sig.get('current_price', 0))
if w_data and w_data.get('data_available', False) and 'trading_signal' in w_data:
signal = w_data['trading_signal']
action = signal.get('action', 'HOLD')
confidence = float(signal.get('confidence', 0.5))
dynamic_impact = SystemLimits.L3_WHALE_IMPACT_MAX * confidence
if action == 'BUY': whale_points = dynamic_impact
elif action == 'SELL': whale_points = -dynamic_impact
except Exception: pass
news_points = 0.0
try:
if news_fetcher and senti_analyzer:
n_data = await news_fetcher.get_news(symbol)
summary_text = n_data.get('summary', '')
if "No specific news" not in summary_text:
sent = senti_analyzer.polarity_scores(summary_text)
compound_score = sent['compound']
news_points = compound_score * SystemLimits.L3_NEWS_IMPACT_MAX
except Exception: pass
mc_a_points = 0.0
try:
raw_mc_a = await ml_processor.run_advanced_monte_carlo(symbol, '1h')
mc_a_points = max(-SystemLimits.L3_MC_ADVANCED_MAX, min(SystemLimits.L3_MC_ADVANCED_MAX, raw_mc_a))
except Exception: pass
final_score = l2_score + whale_points + news_points + mc_a_points
sig['whale_score'] = whale_points
sig['news_score'] = news_points
sig['mc_advanced_score'] = mc_a_points
sig['final_total_score'] = final_score
final_candidates.append(sig)
final_candidates.sort(key=lambda x: x['final_total_score'], reverse=True)
approved_signals = []
header = (f"{'SYM':<9} | {'TYPE':<10} | {'L2':<5} | {'TITAN':<5} | {'PATT':<5} | "
f"{'WHALE':<6} | {'MC(A)':<6} | {'FINAL':<6} | {'ORACLE':<6} | {'STATUS'}")
log_and_print("-" * 115)
log_and_print(header)
log_and_print("-" * 115)
for sig in final_candidates:
symbol = sig['symbol']
decision = await ml_processor.consult_oracle(sig)
action = decision.get('action', 'WAIT')
oracle_conf = decision.get('confidence', 0.0)
target_class = decision.get('target_class', '')
status_str = "WAIT ๐ด"
if action == 'WATCH' or action == 'BUY':
status_str = f"โ
{target_class}"
sig.update(decision)
approved_signals.append(sig)
l2_hybrid = sig.get('enhanced_final_score', 0.0)
titan_d = sig.get('titan_score', 0.0)
patt_d = sig.get('patterns_score', 0.0)
whale_d = sig.get('whale_score', 0.0)
mca_d = sig.get('mc_advanced_score', 0.0)
final_d = sig.get('final_total_score', 0.0)
strat_type = sig.get('strategy_type', 'N/A')
log_and_print(
f"{symbol:<9} | "
f"{strat_type:<10} | "
f"{l2_hybrid:.2f} | "
f"{titan_d:.2f} | "
f"{patt_d:.2f} | "
f"{whale_d:+.2f} | "
f"{mca_d:+.2f} | "
f"{final_d:.2f} | "
f"{oracle_conf:.2f} | "
f"{status_str}"
)
if approved_signals:
log_and_print("-" * 115)
log_and_print(f" [4/5] ๐ฏ L4 Sniper -> ๐๏ธ Governance -> ๐ฐ Portfolio ({len(approved_signals)} candidates)...")
tm_log_buffer = StringIO()
with redirect_stdout(tm_log_buffer), redirect_stderr(tm_log_buffer):
await trade_manager.select_and_execute_best_signal(approved_signals)
tm_logs = tm_log_buffer.getvalue()
for line in tm_logs.splitlines():
if line.strip(): log_and_print(line.strip())
else:
log_and_print(" -> ๐ No candidates approved by Oracle for Sniper check.")
gc.collect()
sys_state.set_cycle_end(logs=log_buffer.getvalue())
except Exception as e:
logger.error(f"โ [Cycle ERROR] {e}")
traceback.print_exc()
sys_state.set_cycle_end(error=e, logs=log_buffer.getvalue())
# ------------------------------------------------------------------------------
# Handlers
# ------------------------------------------------------------------------------
async def trigger_training_cycle():
if adaptive_hub: return f"๐ค Adaptive System: {adaptive_hub.get_status()}"
return "โ ๏ธ System not ready."
async def trigger_strategic_backtest():
if not BACKTEST_AVAILABLE: return "โ ๏ธ Backtest Engine not found."
if trade_manager and len(trade_manager.open_positions) > 0: return "โ Active trades exist."
if sys_state.training_running: return "โ ๏ธ Running."
async def _run_bg_task():
sys_state.training_running = True
sys_state.training_status_msg = "๐งช Strategic Backtest Running..."
try:
logger.info("๐งช [Manual] Starting Strategic Backtest...")
await run_strategic_optimization_task()
if adaptive_hub: await adaptive_hub.initialize()
logger.info("โ
[Manual] Backtest Complete.")
except Exception as e: logger.error(f"โ Backtest Failed: {e}")
finally:
sys_state.training_running = False
sys_state.training_status_msg = adaptive_hub.get_status() if adaptive_hub else "Ready"
asyncio.create_task(_run_bg_task())
return "๐งช Strategic Backtest Started."
async def manual_close_current_trade():
if not trade_manager.open_positions: return "โ ๏ธ No trade."
symbol = list(trade_manager.open_positions.keys())[0]
await trade_manager.force_exit_by_manager(symbol, reason="MANUAL_UI")
return f"โ
Closed {symbol}."
async def reset_history_handler():
if trade_manager.open_positions: return "โ ๏ธ Close active trades first."
current_state = await r2.get_portfolio_state_async()
preserved_capital = current_state.get('current_capital_usd', INITIAL_CAPITAL)
await r2.reset_all_stats_async()
if trade_manager and trade_manager.smart_portfolio:
sp = trade_manager.smart_portfolio
sp.state["current_capital"] = preserved_capital
sp.state["session_start_balance"] = preserved_capital
sp.state["allocated_capital_usd"] = 0.0
sp.state["daily_net_pnl"] = 0.0
sp.state["is_trading_halted"] = False
await sp._save_state_to_r2()
return f"โ
History Cleared. Capital Preserved at ${preserved_capital:.2f}"
async def reset_capital_handler():
if trade_manager.open_positions: return "โ ๏ธ Close active trades first."
if trade_manager and trade_manager.smart_portfolio:
sp = trade_manager.smart_portfolio
sp.state["current_capital"] = INITIAL_CAPITAL
sp.state["session_start_balance"] = INITIAL_CAPITAL
sp.state["allocated_capital_usd"] = 0.0
sp.state["daily_net_pnl"] = 0.0
sp.state["is_trading_halted"] = False
await sp._save_state_to_r2()
return f"โ
Capital Reset to ${INITIAL_CAPITAL} (History Kept)"
async def reset_diagnostics_handler():
await r2.reset_diagnostic_stats_async()
return "โ
Diagnostic Matrix Reset."
async def reset_guardians_handler():
await r2.reset_guardian_stats_async()
if trade_manager: trade_manager.ai_stats = await r2.get_guardian_stats_async()
return "โ
Guardian Stats Reset."
async def toggle_auto_pilot(enable):
sys_state.auto_pilot = enable
return f"Auto-Pilot: {enable}"
async def run_cycle_from_gradio():
if sys_state.cycle_running: return "Busy."
asyncio.create_task(run_unified_cycle())
return "๐ Launched."
# ------------------------------------------------------------------------------
# UI Updates
# ------------------------------------------------------------------------------
async def check_live_pnl_and_status(selected_view="Dual-Core (Hybrid)"):
empty_chart = go.Figure()
empty_chart.update_layout(template="plotly_dark", paper_bgcolor="#0b0f19", plot_bgcolor="#0b0f19", xaxis={'visible':False}, yaxis={'visible':False})
wl_df_empty = pd.DataFrame(columns=["Coin", "Score"])
diag_df_empty = pd.DataFrame(columns=["Model", "Wins", "Losses", "PnL (USD)"])
type_df_empty = pd.DataFrame(columns=["Coin Type", "Wins", "Losses", "Profitability"])
if not sys_state.ready:
return "Initializing...", "...", empty_chart, "0.0", "0.0", "0.0", "0.0", "0.0%", wl_df_empty, diag_df_empty, type_df_empty, "Loading...", "Loading...", "Loading..."
try:
sp = trade_manager.smart_portfolio
equity = sp.state.get('current_capital', 10.0)
allocated = sp.state.get('allocated_capital_usd', 0.0)
free_cap = max(0.0, equity - allocated)
daily_pnl = sp.state.get('daily_net_pnl', 0.0)
is_halted = sp.state.get('is_trading_halted', False)
symbol = None; entry_p = 0.0; tp_p = 0.0; sl_p = 0.0; curr_p = 0.0; pnl_pct = 0.0; pnl_val_unrealized = 0.0
active_trade_info = ""
trade_dur_str = "--:--:--"
if trade_manager.open_positions:
symbol = list(trade_manager.open_positions.keys())[0]
trade = trade_manager.open_positions[symbol]
entry_p = float(trade.get('entry_price', 0.0))
tp_p = float(trade.get('tp_price', 0.0))
sl_p = float(trade.get('sl_price', 0.0))
trade_dur_str = calculate_duration_str(trade.get('entry_time'))
decision_data = trade.get('decision_data', {})
gov_grade = decision_data.get('governance_grade', 'N/A')
gov_score = decision_data.get('governance_score', 0.0)
sys_conf = decision_data.get('system_confidence', 0.0)
strat_type = trade.get('strategy_type', 'NORMAL')
grade_color = "#ccc"
if gov_grade == "ULTRA": grade_color = "#ff00ff"
elif gov_grade == "STRONG": grade_color = "#00ff00"
elif gov_grade == "NORMAL": grade_color = "#00e5ff"
elif gov_grade == "WEAK": grade_color = "#ffff00"
elif gov_grade == "REJECT": grade_color = "#ff0000"
curr_p = await data_manager.get_latest_price_async(symbol)
if curr_p > 0 and entry_p > 0:
pnl_pct = ((curr_p - entry_p) / entry_p) * 100
size = float(trade.get('entry_capital', 0.0))
pnl_val_unrealized = size * (pnl_pct / 100)
active_trade_info = f"""
โฑ๏ธ Time: {trade_dur_str}
๐๏ธ Grade: {gov_grade} ({gov_score:.1f})
๐ท๏ธ Type: {strat_type}
"""
virtual_equity = equity + pnl_val_unrealized
active_pnl_color = "#00ff00" if pnl_val_unrealized >= 0 else "#ff0000"
portfolio = await r2.get_portfolio_state_async()
total_t = portfolio.get('total_trades', 0)
wins = portfolio.get('winning_trades', 0)
losses = portfolio.get('losing_trades', 0)
if losses == 0 and total_t > 0: losses = total_t - wins
tot_prof = portfolio.get('total_profit_usd', 0.0)
tot_loss = portfolio.get('total_loss_usd', 0.0)
net_prof = tot_prof - tot_loss
win_rate = (wins / total_t * 100) if total_t > 0 else 0.0
halt_status = "HALTED" if is_halted else "ACTIVE"
wallet_md = f"""
๐ผ Institutional Portfolio
${virtual_equity:,.2f}
({pnl_val_unrealized:+,.2f} USD)
| Allocated: | ${allocated:.2f} |
| Free Cap: | ${free_cap:.2f} |
| Daily PnL: | ${daily_pnl:+.2f} |
๐ฆ
Mood: {sp.market_trend}
๐ก๏ธ Status: {halt_status}
{active_trade_info}
"""
# Guardian Stats
key_map = {
"Dual-Core (Hybrid)": "hybrid",
"Hydra: Crash (Panic)": "crash",
"Hydra: Giveback (Profit)": "giveback",
"Hydra: Stagnation (Time)": "stagnation"
}
target_key = key_map.get(selected_view, "hybrid")
stats_data = trade_manager.ai_stats.get(target_key, {"total":0, "good":0, "saved":0.0, "missed":0.0})
tot_ds = stats_data['total']
ds_acc = (stats_data['good'] / tot_ds * 100) if tot_ds > 0 else 0.0
# โ
RESTORED MISSED ROW
history_md = f"""
๐ Performance
| Trades: | {total_t} |
| Win Rate: | {win_rate:.1f}% |
| Wins: | {wins} (+${tot_prof:,.2f}) |
| Losses: | {losses} (-${tot_loss:,.2f}) |
| Net: | ${net_prof:,.2f} |
๐ก๏ธ Guard IQ ({target_key})
| Interventions: | {tot_ds} |
| Accuracy: | {ds_acc:.1f}% |
| Saved: | ${stats_data['saved']:.2f} |
| Missed: | ${stats_data['missed']:.2f} |
"""
fast_learn_prog = "0/100"
if adaptive_hub:
if hasattr(adaptive_hub, 'get_learning_progress'):
fast_learn_prog = adaptive_hub.get_learning_progress()
metrics = scheduler.get_status_metrics() if scheduler else {}
sch_w_time = metrics.get("weekly_timer", "Wait")
sch_m_time = metrics.get("monthly_timer", "Wait")
neural_md = f"""
๐ง Neural Cycles
| โก Fast Learner: |
{fast_learn_prog} |
| ๐
Weekly Tune: |
{sch_w_time} |
| ๐๏ธ Monthly Evo: |
{sch_m_time} |
"""
# Diagnostic Matrix
diag_data = await r2.get_diagnostic_stats_async()
diag_list = []
ordered_models = ["Titan", "Patterns", "Oracle", "Sniper", "MonteCarlo_L", "Governance"]
for m in ordered_models:
stats = diag_data.get(m, {"wins": 0, "losses": 0, "pnl": 0.0})
pnl_val = stats['pnl']
color = "#00ff00" if pnl_val >= 0 else "#ff0000"
pnl_str = f"${pnl_val:+.2f}"
diag_list.append([m, stats['wins'], stats['losses'], pnl_str])
diag_df = pd.DataFrame(diag_list, columns=["Model", "Wins", "Losses", "PnL (USD)"])
# Type Stats DataFrame
type_stats_list = []
if trade_manager and hasattr(trade_manager, 'type_stats'):
for t_name, t_data in trade_manager.type_stats.items():
name_clean = t_name.replace("_", " ")
wins = t_data.get('wins', 0)
losses = t_data.get('losses', 0)
prof = t_data.get('profit_usd', 0.0)
loss_val = t_data.get('loss_usd', 0.0)
profitability_html = format_pnl_split(prof, loss_val)
type_stats_list.append([name_clean, wins, losses, profitability_html])
type_df = pd.DataFrame(type_stats_list, columns=["Coin Type", "Wins", "Losses", "Profitability"])
wl_data = [[k, f"{v.get('final_total_score',0):.2f}"] for k, v in trade_manager.watchlist.items()]
wl_df = pd.DataFrame(wl_data, columns=["Coin", "Score"])
status_txt = sys_state.last_cycle_logs
status_line = f"Cycle: {'RUNNING' if sys_state.cycle_running else 'IDLE'} | Auto-Pilot: {'ON' if sys_state.auto_pilot else 'OFF'}"
fig = empty_chart
if symbol and curr_p > 0:
ohlcv = await data_manager.get_latest_ohlcv(symbol, '5m', 120)
if ohlcv:
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
fig = go.Figure(data=[go.Candlestick(
x=df['datetime'], open=df['open'], high=df['high'], low=df['low'], close=df['close'],
increasing_line_color='#00ff00', decreasing_line_color='#ff0000', name=symbol
)])
if entry_p > 0:
fig.add_hline(y=entry_p, line_dash="dash", line_color="white", annotation_text="ENTRY", annotation_position="top left")
if tp_p > 0:
fig.add_hline(y=tp_p, line_color="#00ff00", line_width=2, annotation_text="TP", annotation_position="top left")
if sl_p > 0:
fig.add_hline(y=sl_p, line_color="#ff0000", line_width=2, annotation_text="SL", annotation_position="bottom left")
fig.update_layout(
template="plotly_dark",
paper_bgcolor="#0b0f19",
plot_bgcolor="#0b0f19",
margin=dict(l=0, r=40, t=30, b=0),
height=400,
xaxis_rangeslider_visible=False,
title=dict(text=f"{symbol} (Long) | PnL: {pnl_pct:+.2f}%", font=dict(color="white"))
)
train_status = sys_state.training_status_msg
if sys_state.training_running: train_status = "๐งช Backtest Running..."
return (status_txt, status_line, fig, f"{curr_p:.6f}", f"{entry_p:.6f}", f"{tp_p:.6f}", f"{sl_p:.6f}",
f"{pnl_pct:+.2f}%", wl_df, diag_df, type_df, wallet_md, history_md, neural_md)
except Exception:
traceback.print_exc()
return "Error", "Error", empty_chart, "0", "0", "0", "0", "0%", wl_df_empty, diag_df_empty, type_df_empty, "Err", "Err", "Err"
# ------------------------------------------------------------------------------
# Gradio UI Construction
# ------------------------------------------------------------------------------
def create_gradio_ui():
custom_css = ".gradio-container {background:#0b0f19} .dataframe {background:#1a1a1a!important} .html-box {min-height:180px}"
with gr.Blocks(title="Titan V61.4 (Full Neural Dashboard)") as demo:
gr.HTML(f"")
gr.Markdown("# ๐ Titan V61.4 (Cybernetic: Dual-Type Engine)")
with gr.Row():
with gr.Column(scale=3):
live_chart = gr.Plot(label="Chart")
with gr.Row():
t_price = gr.Textbox(label="Price", interactive=False)
t_pnl = gr.Textbox(label="PnL %", interactive=False)
with gr.Row():
t_entry = gr.Textbox(label="Entry", interactive=False)
t_tp = gr.Textbox(label="TP", interactive=False)
t_sl = gr.Textbox(label="SL", interactive=False)
with gr.Column(scale=1):
wallet_out = gr.HTML(label="Smart Wallet", elem_classes="html-box")
neural_out = gr.HTML(label="Neural Cycles", elem_classes="html-box")
# Type Stats Table
gr.Markdown("### ๐ Opportunity Types")
type_stats_out = gr.Dataframe(
headers=["Coin Type", "Wins", "Losses", "Profitability"],
datatype=["str", "number", "number", "html"],
interactive=False,
label="Type Performance"
)
gr.Markdown("### ๐ต๏ธ Diagnostic Matrix")
diagnostic_out = gr.Dataframe(
headers=["Model", "Wins", "Losses", "PnL (USD)"],
datatype=["str", "number", "number", "html"],
interactive=False,
label="Model Performance"
)
with gr.Row():
btn_reset_diag = gr.Button("๐งน Reset Matrix", size="sm", variant="secondary")
btn_reset_guard = gr.Button("๐ก๏ธ Reset Guardians", size="sm", variant="secondary")
gr.HTML("
")
stats_dd = gr.Dropdown([
"Dual-Core (Hybrid)",
"Hydra: Crash (Panic)",
"Hydra: Giveback (Profit)",
"Hydra: Stagnation (Time)"
], value="Dual-Core (Hybrid)", label="View Guard Stats")
history_out = gr.HTML(label="Stats", elem_classes="html-box")
watchlist_out = gr.DataFrame(label="Watchlist")
gr.HTML("
")
with gr.Row():
with gr.Column(scale=1):
auto_pilot = gr.Checkbox(label="โ๏ธ Auto-Pilot", value=True)
with gr.Row():
btn_run = gr.Button("๐ Scan", variant="primary")
btn_close = gr.Button("๐จ Close", variant="stop")
with gr.Row():
btn_train = gr.Button("๐ค Status", variant="secondary")
btn_backtest = gr.Button("๐งช Run Strategic Backtest", variant="secondary")
with gr.Row():
btn_history_reset = gr.Button("๐๏ธ Clear History", variant="secondary")
btn_cap_reset = gr.Button("๐ฐ Reset Capital", variant="secondary")
status = gr.Markdown("Init...")
alert = gr.Textbox(label="Alerts", interactive=False)
with gr.Column(scale=3):
logs = gr.Textbox(label="Logs", lines=14, autoscroll=True, elem_classes="log-box", type="text")
gr.HTML("")
btn_run.click(fn=run_cycle_from_gradio, outputs=alert)
btn_close.click(fn=manual_close_current_trade, outputs=alert)
btn_history_reset.click(fn=reset_history_handler, outputs=alert)
btn_cap_reset.click(fn=reset_capital_handler, outputs=alert)
btn_train.click(fn=trigger_training_cycle, outputs=alert)
btn_backtest.click(fn=trigger_strategic_backtest, outputs=alert)
auto_pilot.change(fn=toggle_auto_pilot, inputs=auto_pilot, outputs=alert)
btn_reset_diag.click(fn=reset_diagnostics_handler, outputs=alert)
btn_reset_guard.click(fn=reset_guardians_handler, outputs=alert)
gr.Timer(3).tick(fn=check_live_pnl_and_status, inputs=stats_dd,
outputs=[logs, status, live_chart, t_price, t_entry, t_tp, t_sl, t_pnl, watchlist_out, diagnostic_out, type_stats_out, wallet_out, history_out, neural_out])
return demo
fast_api_server = FastAPI(lifespan=lifespan)
gradio_dashboard = create_gradio_ui()
app = gr.mount_gradio_app(app=fast_api_server, blocks=gradio_dashboard, path="/")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)