Gli 8 servizi v6

async for msg in binance_ws.stream(klines_universe):     # real-time
    queue.put(normalize_kline(msg))

every 60s:                                               # REST pollers
    for src in [funding, open_interest, long_short]:
        for r in fetch(src): queue.put(normalize(r))
every 300s:
    for src in [fear_greed, dxy, stablecoin_supply]:
        for r in fetch(src): queue.put(normalize(r))

while batch := queue.drain(max=500):
    server_log.upsert(batch)        # delega la persistenza

def upsert(batch):
    for table, rows in partition_by_table(batch).items():
        execute_values(
            f"INSERT INTO {table} (...) VALUES %s "
            "ON CONFLICT (pk) DO UPDATE SET ...", rows, page_size=1000)

daily 04:00:                              # retention 180gg
    for t in candle_tables:
        DELETE FROM t WHERE open_time < now_ms() - days(180)

every 60s:
    for sym in universe:                       # 200+ asset
        c = candles_1m.last_n(sym, 60)
        if ret(c, 5m) >  PUMP_PCT:     emit("pump", sym, sev="warning")
        if ret(c, 5m) < -DUMP_PCT:     emit("dump", sym, sev="warning")
        if zscore(c.volume, 60) > VOL_Z: emit("volume_spike", sym)
        if funding_z21(sym) >= FUND_Q:   emit("funding_extreme", sym)
        if liquidity_drop(sym) > LIQ_PCT: emit("liquidity_drop", sym)

every 1h:
    for sym in universe:
        for tf in [1h, 4h, 1d]:
            c = candles[tf].last_n(sym, 250)
            feats = {
              "rsi14": rsi(c,14), "ema": ema(c,[9,21,50,200]),
              "atr14": atr(c,14), "ret": returns(c),
              "boll": bollinger(c,20,2), "vwap": vwap(c),
              "beta": beta(c, btc), "funding_z21": z(funding(sym),21),
              "funding_slope": slope(funding(sym)), "oi": oi(sym),
            }
            feature_store.put(sym, tf, feats)

every 1h:
    for sym in universe:
        scores = {m.name: m.evaluate(sym, features[sym]) for m in MODULES}  # 15 moduli
        agg = sum(scores[m.name]*WEIGHT[m.cat] for m in MODULES) / sum_weights
        direction = "short" if defensive_edge(sym) else "long"   # edge Funding Q90
        analyz_signals.insert(sym, signal_type=top(scores),
                              direction=direction, score=agg, payload=scores)

monthly:
    WEIGHT = walk_forward_revalidate(history)   # auto-validazione pesi

on new analyz_signals:
    for sig in analyz_signals.fresh():
        if not passes_filters(sig.symbol): continue   # universe filter
        if sig.score >= ENTRY_TH and not open(sig.symbol):
            size = position_size(equity, RISK_PCT)
            decisions.insert("buy", sig); paper_open(sig.symbol, size)

    for pos in paper_positions.open():
        if exit_rule(pos):              # TP / SL / holding max / decay score
            decisions.insert("close", pos, reason=exit_reason(pos))
            paper_close(pos)

every 5min:                              # health
    h = collect_heartbeats()
    if stale(h): quality_reports.insert("health", status="critical", details=h)

daily:
    for trade in closed_trades(yesterday):
        attr = attribute_pnl(trade)          # perche' ha perso/guadagnato
        quality_reports.insert("loss_analysis", scope=trade.id, details=attr)
    missed = scan_missed_opportunities()
    quality_reports.insert("opportunity", details=missed)

loop:                                  # container isolato
    if now - last_heartbeat > 5min:    trip("heartbeat")
    if drawdown()       >= 0.20:       trip("drawdown_20")
    if daily_loss()     >= 0.07:       trip("daily_loss_7")
    if max_position_pct() >= 0.10:     trip("position_10")
    if orders_per_min() >  LIMIT:      trip("order_rate")
    if btc_move_10min() <= -0.05:      trip("flash_crash_btc")
    if exchange_error():               trip("exchange_error")
    if quality_critical():             trip("quality_critical")

def trip(reason):
    market_sell_all(); emergency_actions.insert(reason); halt()  # restart MANUALE