lignarr/Trainer-PPO
0
1"""2app.py — Enhanced Telegram Command Center for PPO Training Space.3Full control: train, backtest, validate, model management, logs, system info.4"""5 6import os, subprocess, threading, collections, shutil, json, glob, time7from datetime import datetime8from flask import Flask, request as flask_request, jsonify9 10app = Flask(__name__)11 12SPACE_NAME = os.environ.get("SPACE_NAME", "PPO Trainer")13current_process = None14process_mode = None15process_start_time = None16log_buffer = collections.deque(maxlen=300)17 18# ═══════════════════════════════════════════════════════════19# MENUS — Deep, multi-level Telegram control20# ═══════════════════════════════════════════════════════════21def menu_main():22 return {"inline_keyboard": [23 [{"text": "🚀 Pipelines", "callback_data": "menu_pipes"}],24 [{"text": "📊 Monitoring", "callback_data": "menu_mon"}],25 [{"text": "📦 Models", "callback_data": "menu_models"}],26 [{"text": "⚙️ Controls", "callback_data": "menu_ctrl"}],27 [{"text": "🧹 Maintenance", "callback_data": "menu_maint"}],28 ]}29 30def menu_pipes():31 return {"inline_keyboard": [32 [{"text": "🔄 Full Pipeline (Train→Backtest→Validate)", "callback_data": "run_full"}],33 [{"text": "📥 Extract Data", "callback_data": "run_extract"}],34 [{"text": "🧠 Train Only", "callback_data": "run_train"}],35 [{"text": "📊 Backtest Only", "callback_data": "run_backtest"}],36 [{"text": "✅ Validate Only", "callback_data": "run_validate"}],37 [{"text": "⬅️ Back", "callback_data": "menu_main"}],38 ]}39 40def menu_mon():41 return {"inline_keyboard": [42 [{"text": "📋 Logs (last 30)", "callback_data": "mon_logs_30"}],43 [{"text": "📋 Logs (last 100)", "callback_data": "mon_logs_100"}],44 [{"text": "📈 Training Metrics", "callback_data": "mon_metrics"}],45 [{"text": "💾 Disk & Files", "callback_data": "mon_disk"}],46 [{"text": "🖥️ System Info", "callback_data": "mon_system"}],47 [{"text": "📊 Backtest Results", "callback_data": "mon_backtest"}],48 [{"text": "⬅️ Back", "callback_data": "menu_main"}],49 ]}50 51def menu_models():52 return {"inline_keyboard": [53 [{"text": "📦 List All Models", "callback_data": "mdl_list"}],54 [{"text": "🏆 Best Model Info", "callback_data": "mdl_best"}],55 [{"text": "✅ Deployment-Ready?", "callback_data": "mdl_deploy_status"}],56 [{"text": "🗑️ Clear Checkpoints", "callback_data": "mdl_clear_ckpt"}],57 [{"text": "🗑️ Clear ALL Models", "callback_data": "mdl_clear_all"}],58 [{"text": "⬅️ Back", "callback_data": "menu_main"}],59 ]}60 61def menu_ctrl():62 return {"inline_keyboard": [63 [{"text": "ℹ️ Detailed Status", "callback_data": "ctrl_status"}],64 [{"text": "🛑 Stop Process", "callback_data": "ctrl_stop"}],65 [{"text": "🔁 Restart Space", "callback_data": "ctrl_restart"}],66 [{"text": "⬅️ Back", "callback_data": "menu_main"}],67 ]}68 69def menu_maint():70 return {"inline_keyboard": [71 [{"text": "🗑️ Clear Logs", "callback_data": "maint_clear_logs"}],72 [{"text": "🗑️ Clear Data Cache", "callback_data": "maint_clear_data"}],73 [{"text": "🗑️ Clear Results", "callback_data": "maint_clear_results"}],74 [{"text": "🔧 Re-extract Features", "callback_data": "run_features"}],75 [{"text": "📏 Check Data Integrity", "callback_data": "maint_check_data"}],76 [{"text": "⬅️ Back", "callback_data": "menu_main"}],77 ]}78 79# ═══════════════════════════════════════════════════════════80# PROCESS MANAGEMENT81# ═══════════════════════════════════════════════════════════82def run_pipeline(mode):83 global current_process, process_mode, process_start_time84 if current_process is not None and current_process.poll() is None:85 return False86 log_buffer.clear()87 process_mode = mode88 process_start_time = datetime.now()89 90 # Map mode to the right command91 cmd_map = {92 "full": ["python", "-u", "train_and_validate.py", "--mode", "full"],93 "train": ["python", "-u", "train_and_validate.py", "--mode", "train"],94 "backtest": ["python", "-u", "train_and_validate.py", "--mode", "backtest"],95 "validate": ["python", "-u", "train_and_validate.py", "--mode", "validate"],96 "extract": ["python", "-u", "main.py", "--mode", "extract"],97 "features": ["python", "-u", "main.py", "--mode", "features"],98 }99 cmd = cmd_map.get(mode, ["python", "-u", "main.py", "--mode", mode])100 101 def worker():102 global current_process103 try:104 current_process = subprocess.Popen(105 cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,106 text=True, bufsize=1)107 for line in current_process.stdout:108 log_buffer.append(line.rstrip())109 current_process.wait()110 log_buffer.append(f"--- {mode} exited: {current_process.returncode} ---")111 except Exception as e:112 log_buffer.append(f"ERROR: {e}")113 threading.Thread(target=worker, daemon=True).start()114 return True115 116# ═══════════════════════════════════════════════════════════117# MONITORING HELPERS118# ═══════════════════════════════════════════════════════════119def get_logs(n=30):120 lines = list(log_buffer)[-n:]121 if not lines:122 return "<i>No logs yet.</i>"123 return "\n".join(l.replace("<", "<").replace(">", ">")[:120] for l in lines)124 125def get_training_metrics():126 log_dir = "logs"127 if not os.path.isdir(log_dir):128 return "<i>No log directory.</i>"129 logs = sorted(glob.glob(os.path.join(log_dir, "*.log")), key=os.path.getmtime, reverse=True)130 if not logs:131 return "<i>No log files found.</i>"132 try:133 with open(logs[0], "r", errors="ignore") as f:134 lines = f.readlines()[-50:]135 # Find key metrics from recent lines136 info = {"balance": "?", "win_rate": "?", "profit_factor": "?", "trades": "?", "timesteps": "?"}137 for line in reversed(lines):138 l = line.lower()139 for key in info:140 if key in l and info[key] == "?":141 parts = line.split("|") if "|" in line else [line]142 for p in parts:143 if key in p.lower():144 val = p.split(":")[-1].strip() if ":" in p else p.strip()145 info[key] = val[:30]146 break147 return (f"📈 <b>Training Metrics</b>\n<i>{os.path.basename(logs[0])}</i>\n\n"148 f"💰 Balance: <b>{info['balance']}</b>\n"149 f"🎯 Win Rate: <b>{info['win_rate']}</b>\n"150 f"📊 PF: <b>{info['profit_factor']}</b>\n"151 f"📊 Trades: <b>{info['trades']}</b>\n"152 f"⏱ Steps: <b>{info['timesteps']}</b>")153 except Exception as e:154 return f"<i>Error: {e}</i>"155 156def get_backtest_results():157 path = os.path.join("results", "backtest_results.json")158 if not os.path.exists(path):159 return "<i>No backtest results yet. Run a backtest first.</i>"160 try:161 with open(path) as f:162 data = json.load(f)163 m = data.get("metrics", {})164 return (f"📊 <b>Last Backtest Results</b>\n\n"165 f"💰 Final Balance: <b>${m.get('Final Balance', '?')}</b>\n"166 f"🎯 Win Rate: <b>{m.get('Win Rate (%)', '?')}%</b>\n"167 f"📊 Profit Factor: <b>{m.get('Profit Factor', '?')}</b>\n"168 f"📉 Max Drawdown: <b>{m.get('Max Drawdown (%)', '?')}%</b>\n"169 f"📈 Sharpe: <b>{m.get('Annualized Sharpe', m.get('Sharpe Ratio', '?'))}</b>\n"170 f"📊 Total Trades: <b>{m.get('Total Trades', '?')}</b>\n"171 f"💵 Expectancy: <b>${m.get('Expectancy', '?')}</b>")172 except Exception as e:173 return f"<i>Error reading results: {e}</i>"174 175def get_disk_info():176 total, used, free = shutil.disk_usage("/")177 text = (f"💾 <b>Disk</b>\nTotal: {total//(1024**3)}GB | "178 f"Used: {used//(1024**3)}GB | Free: {free//(1024**3)}GB\n\n📁 <b>Key Files:</b>\n")179 checks = [("data/eurusd_features.csv", "Features"), ("data/eurusd_master_dataset.csv", "Master Data"),180 ("data/norm_params.pkl", "Norm Params"), ("models/best_model.zip", "Best Model"),181 ("models/deployment_ready_model.zip", "Deploy-Ready Model"),182 ("results/backtest_results.json", "Backtest Results")]183 for path, label in checks:184 if os.path.exists(path):185 sz = os.path.getsize(path) / (1024*1024)186 text += f" ✅ {label}: <b>{sz:.1f}MB</b>\n"187 else:188 text += f" ❌ {label}: <i>missing</i>\n"189 return text190 191def get_system_info():192 import sys193 lines = [f"🖥️ <b>System Info</b>", f"CPUs: <b>{os.cpu_count()}</b>",194 f"Python: <b>{sys.version.split()[0]}</b>",195 f"Time: <b>{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</b>"]196 try:197 with open("/proc/meminfo") as f:198 for line in f:199 if line.startswith("MemTotal:"):200 lines.append(f"RAM Total: <b>{int(line.split()[1])//1024}MB</b>")201 elif line.startswith("MemAvailable:"):202 lines.append(f"RAM Free: <b>{int(line.split()[1])//1024}MB</b>")203 except:204 pass205 try:206 load1, load5, load15 = os.getloadavg()207 lines.append(f"Load: <b>{load1:.2f} / {load5:.2f} / {load15:.2f}</b>")208 except:209 pass210 return "\n".join(lines)211 212def list_models():213 model_dir = "models"214 if not os.path.isdir(model_dir):215 return "<i>No models directory.</i>"216 files = sorted(os.listdir(model_dir))217 if not files:218 return "<i>No models found.</i>"219 text = "📦 <b>Models</b>\n\n"220 for f in files:221 fp = os.path.join(model_dir, f)222 sz = os.path.getsize(fp) / (1024*1024)223 icon = "🏆" if "best" in f else ("✅" if "deploy" in f else "📄")224 text += f" {icon} {f} ({sz:.1f}MB)\n"225 return text226 227def get_best_model_info():228 path = os.path.join("models", "best_model.zip")229 if not os.path.exists(path):230 return "<i>No best model yet.</i>"231 sz = os.path.getsize(path) / (1024*1024)232 mtime = datetime.fromtimestamp(os.path.getmtime(path))233 return (f"🏆 <b>Best Model</b>\n\n"234 f"Size: <b>{sz:.1f}MB</b>\n"235 f"Modified: <b>{mtime.strftime('%Y-%m-%d %H:%M:%S')}</b>")236 237def check_deploy_ready():238 path = os.path.join("models", "deployment_ready_model.zip")239 if os.path.exists(path):240 sz = os.path.getsize(path) / (1024*1024)241 mtime = datetime.fromtimestamp(os.path.getmtime(path))242 return (f"✅ <b>Deployment-Ready Model EXISTS</b>\n\n"243 f"Size: <b>{sz:.1f}MB</b>\n"244 f"Validated: <b>{mtime.strftime('%Y-%m-%d %H:%M')}</b>\n\n"245 "This model passed all validation gates.")246 return "❌ <b>No deployment-ready model.</b>\n\nRun the full pipeline to generate one."247 248def check_data_integrity():249 features = "data/eurusd_features.csv"250 if not os.path.exists(features):251 return "❌ <b>Features CSV missing!</b> Run data extraction."252 try:253 import pandas as pd254 df = pd.read_csv(features, nrows=5)255 total = sum(1 for _ in open(features)) - 1256 return (f"✅ <b>Data Integrity Check</b>\n\n"257 f"Rows: <b>{total:,}</b>\n"258 f"Columns: <b>{len(df.columns)}</b>\n"259 f"Features: <b>{', '.join(df.columns[:8].tolist())}...</b>")260 except Exception as e:261 return f"⚠️ <b>Data issue:</b> {e}"262 263# ═══════════════════════════════════════════════════════════264# WEBHOOK HANDLER265# ═══════════════════════════════════════════════════════════266@app.route("/webhook", methods=["POST"])267def webhook():268 data = flask_request.get_json(silent=True) or {}269 270 msg = data.get("message", {})271 if msg:272 chat_id = msg.get("chat", {}).get("id")273 text = (msg.get("text") or "").strip()274 if text.startswith("/start"):275 return jsonify({"method": "sendMessage", "chat_id": chat_id,276 "text": (f"🤖 <b>{SPACE_NAME} — Command Center</b>\n\n"277 f"📡 Online | 🕐 {datetime.now().strftime('%H:%M:%S')}\n\n"278 "Full control over training, backtesting, models, and system."),279 "parse_mode": "HTML", "reply_markup": menu_main()})280 elif text.startswith("/logs"):281 return jsonify({"method": "sendMessage", "chat_id": chat_id,282 "text": f"📋 <b>Logs</b>\n\n<pre>{get_logs(30)}</pre>",283 "parse_mode": "HTML", "reply_markup": menu_mon()})284 elif text.startswith("/status"):285 running = current_process is not None and current_process.poll() is None286 s = f"🟢 <b>Running:</b> {process_mode}" if running else "🔴 <b>Idle</b>"287 return jsonify({"method": "sendMessage", "chat_id": chat_id,288 "text": s, "parse_mode": "HTML", "reply_markup": menu_ctrl()})289 elif text.startswith("/models"):290 return jsonify({"method": "sendMessage", "chat_id": chat_id,291 "text": list_models(), "parse_mode": "HTML", "reply_markup": menu_models()})292 elif text.startswith("/train"):293 ok = run_pipeline("full")294 s = "🚀 <b>Full Pipeline Started!</b>" if ok else "⚠️ Already running!"295 return jsonify({"method": "sendMessage", "chat_id": chat_id,296 "text": s, "parse_mode": "HTML", "reply_markup": menu_pipes()})297 elif text.startswith("/backtest"):298 return jsonify({"method": "sendMessage", "chat_id": chat_id,299 "text": get_backtest_results(), "parse_mode": "HTML", "reply_markup": menu_mon()})300 return jsonify({"method": "sendMessage", "chat_id": chat_id,301 "text": "Commands: /start /logs /status /models /train /backtest",302 "parse_mode": "HTML"})303 304 cb = data.get("callback_query", {})305 if cb:306 chat_id = cb["message"]["chat"]["id"]307 msg_id = cb["message"]["message_id"]308 choice = cb.get("data", "")309 310 resp_text, resp_menu = "", menu_main()311 312 # Navigation313 if choice == "menu_main":314 resp_text = f"🤖 <b>{SPACE_NAME}</b>\n🕐 {datetime.now().strftime('%H:%M:%S')}"315 resp_menu = menu_main()316 elif choice == "menu_pipes":317 resp_text = "🚀 <b>Pipeline Control</b>\nSelect a pipeline to run:"318 resp_menu = menu_pipes()319 elif choice == "menu_mon":320 resp_text = "📊 <b>Monitoring</b>"321 resp_menu = menu_mon()322 elif choice == "menu_models":323 resp_text = "📦 <b>Model Management</b>"324 resp_menu = menu_models()325 elif choice == "menu_ctrl":326 resp_text = "⚙️ <b>Controls</b>"327 resp_menu = menu_ctrl()328 elif choice == "menu_maint":329 resp_text = "🧹 <b>Maintenance</b>"330 resp_menu = menu_maint()331 332 # Pipelines333 elif choice in ("run_full", "run_train", "run_backtest", "run_validate", "run_extract", "run_features"):334 mode = choice.replace("run_", "")335 ok = run_pipeline(mode)336 labels = {"full": "Full Pipeline", "train": "Training", "backtest": "Backtest",337 "validate": "Validation", "extract": "Data Extraction", "features": "Feature Engineering"}338 resp_text = f"🚀 <b>{labels.get(mode, mode)} Started!</b>" if ok else "⚠️ <b>Already running!</b>"339 resp_menu = menu_pipes()340 341 # Monitoring342 elif choice == "mon_logs_30":343 resp_text = f"📋 <b>Logs (last 30)</b>\n\n<pre>{get_logs(30)}</pre>"344 resp_menu = menu_mon()345 elif choice == "mon_logs_100":346 resp_text = f"📋 <b>Logs (last 100)</b>\n\n<pre>{get_logs(100)}</pre>"347 resp_menu = menu_mon()348 elif choice == "mon_metrics":349 resp_text = get_training_metrics()350 resp_menu = menu_mon()351 elif choice == "mon_disk":352 resp_text = get_disk_info()353 resp_menu = menu_mon()354 elif choice == "mon_system":355 resp_text = get_system_info()356 resp_menu = menu_mon()357 elif choice == "mon_backtest":358 resp_text = get_backtest_results()359 resp_menu = menu_mon()360 361 # Models362 elif choice == "mdl_list":363 resp_text = list_models()364 resp_menu = menu_models()365 elif choice == "mdl_best":366 resp_text = get_best_model_info()367 resp_menu = menu_models()368 elif choice == "mdl_deploy_status":369 resp_text = check_deploy_ready()370 resp_menu = menu_models()371 elif choice == "mdl_clear_ckpt":372 ckpt_dir = "checkpoints"373 if os.path.isdir(ckpt_dir):374 count = len([f for f in os.listdir(ckpt_dir) if f.endswith(".zip")])375 for f in os.listdir(ckpt_dir):376 if f.endswith(".zip"):377 os.remove(os.path.join(ckpt_dir, f))378 resp_text = f"🗑️ <b>Cleared {count} checkpoints.</b>"379 else:380 resp_text = "ℹ️ No checkpoints to clear."381 resp_menu = menu_models()382 elif choice == "mdl_clear_all":383 model_dir = "models"384 if os.path.isdir(model_dir):385 count = len(os.listdir(model_dir))386 shutil.rmtree(model_dir)387 os.makedirs(model_dir, exist_ok=True)388 resp_text = f"🗑️ <b>Cleared {count} model files.</b>"389 else:390 resp_text = "ℹ️ No models to clear."391 resp_menu = menu_models()392 393 # Controls394 elif choice == "ctrl_status":395 running = current_process is not None and current_process.poll() is None396 if running:397 elapsed = datetime.now() - process_start_time398 recent = list(log_buffer)[-5:]399 recent_text = "\n".join(l.replace("<", "<").replace(">", ">")[:100] for l in recent)400 resp_text = (f"🟢 <b>RUNNING: {process_mode}</b>\n"401 f"Elapsed: <b>{elapsed}</b>\n"402 f"PID: <b>{current_process.pid}</b>\n"403 f"Log lines: <b>{len(log_buffer)}</b>\n\n"404 f"<pre>{recent_text}</pre>")405 else:406 code = current_process.returncode if current_process else "N/A"407 resp_text = f"🔴 <b>Idle</b>\nLast: {process_mode or 'none'} (exit: {code})"408 resp_menu = menu_ctrl()409 elif choice == "ctrl_stop":410 if current_process and current_process.poll() is None:411 current_process.terminate()412 resp_text = "🛑 <b>Process terminated.</b>"413 else:414 resp_text = "ℹ️ Nothing running."415 resp_menu = menu_ctrl()416 elif choice == "ctrl_restart":417 resp_text = "🔁 <b>Use HF Space Settings to restart.</b>\nOr stop+start a pipeline."418 resp_menu = menu_ctrl()419 420 # Maintenance421 elif choice == "maint_clear_logs":422 log_buffer.clear()423 log_dir = "logs"424 if os.path.isdir(log_dir):425 for f in os.listdir(log_dir):426 os.remove(os.path.join(log_dir, f))427 resp_text = "🗑️ <b>All logs cleared.</b>"428 resp_menu = menu_maint()429 elif choice == "maint_clear_data":430 data_dir = "data"431 if os.path.isdir(data_dir):432 for f in os.listdir(data_dir):433 if f.endswith((".csv", ".pkl")):434 os.remove(os.path.join(data_dir, f))435 resp_text = "🗑️ <b>Data cache cleared.</b> Re-extract to rebuild."436 resp_menu = menu_maint()437 elif choice == "maint_clear_results":438 res_dir = "results"439 if os.path.isdir(res_dir):440 shutil.rmtree(res_dir)441 os.makedirs(res_dir, exist_ok=True)442 resp_text = "🗑️ <b>Results cleared.</b>"443 resp_menu = menu_maint()444 elif choice == "maint_check_data":445 resp_text = check_data_integrity()446 resp_menu = menu_maint()447 448 if len(resp_text) > 4000:449 resp_text = resp_text[:3990] + "\n<i>...truncated</i>"450 451 return jsonify({"method": "editMessageText", "chat_id": chat_id,452 "message_id": msg_id, "text": resp_text, "parse_mode": "HTML",453 "reply_markup": resp_menu})454 455 return "ok", 200456 457@app.route("/")458def health():459 return f"{SPACE_NAME} Online", 200460 461if __name__ == "__main__":462 print(f"=== {SPACE_NAME} starting on port 7860 ===")463 app.run(host="0.0.0.0", port=7860)464