Logged every PROGRESS_LOG_INTERVAL_TICKS (default: 60 ticks = ~30 min):
genetic/monitor.py
def log_tick_progress( logger: logging.Logger, gen_num: int, tick_count: int, bots: list[GeneticBot], feed: MarketDataFeed | None = None,): elapsed_hrs = tick_count * TICK_INTERVAL_SECONDS / 3600 active = sum(1 for b in bots if b.account.total_trades > 0) open_pos = sum(b.account.n_open for b in bots) total_trades = sum(b.account.total_trades for b in bots) total_settled = sum(b.account.n_settled for b in bots) # Realized ROI (from settled positions) realized_rois = [b.account.roi_pct for b in bots] best_realized = max(realized_rois) median_realized = sorted(realized_rois)[len(realized_rois) // 2] # Unrealized ROI (estimated from current market prices) if feed: total_rois = [b.account.total_roi_pct(feed) for b in bots] best_total = max(total_rois) median_total = sorted(total_rois)[len(total_rois) // 2]
def compute_generation_stats(bots: list[GeneticBot]) -> dict: """Compute summary stats for a completed generation.""" fitness = [evaluate_fitness(b) for b in bots] trades = [b.account.total_trades for b in bots] win_rates = [b.account.win_rate * 100 for b in bots] sorted_fitness = sorted(fitness) return { "best_roi": sorted_fitness[-1], "median_roi": sorted_fitness[len(sorted_fitness) // 2], "worst_roi": sorted_fitness[0], "mean_roi": sum(fitness) / len(fitness), "mean_trades": sum(trades) / len(trades), "max_trades": max(trades), "mean_win_rate": sum(win_rates) / len(win_rates), "total_settled": sum(b.account.n_settled for b in bots), "active_bots": sum(1 for b in bots if b.account.total_trades > 0), }
from genetic.monitor import setup_logginglogger = setup_logging()# During evolution loopdef custom_monitor(bots, feed): # Track signal type performance signal_performance = {} for bot in bots: sig = bot.params['signal_type'] if sig not in signal_performance: signal_performance[sig] = [] signal_performance[sig].append(bot.account.roi_pct) for sig, rois in signal_performance.items(): avg_roi = sum(rois) / len(rois) logger.info(f"Signal {sig}: avg ROI = {avg_roi:.1f}%")
from flask import Flask, jsonifyimport jsonapp = Flask(__name__)@app.route("/api/latest")def get_latest(): with open("data/evolution/latest.json") as f: pointer = json.load(f) with open(pointer["file"]) as f: data = json.load(f) return jsonify(data["stats"])@app.route("/api/hof")def get_hof(): with open("data/evolution/hall_of_fame.json") as f: return jsonify(json.load(f))if __name__ == "__main__": app.run(port=8000)
def check_evolution_health(stats: dict): """Alert if evolution is unhealthy.""" if stats["active_bots"] < 50: logger.warning( f"LOW ACTIVITY: Only {stats['active_bots']}/100 bots trading. " "Check market availability or reduce filter strictness." ) if stats["total_settled"] < 300: logger.warning( f"LOW SETTLEMENTS: Only {stats['total_settled']} settled trades. " "Extend generation duration or reduce settlement timeout." ) if stats["best_roi"] < -50: logger.error( f"CRITICAL: Best bot has {stats['best_roi']:.1f}% ROI. " "Check for bugs in signal logic or market data issues." )