signal_generator.py

import os
import time

from dotenv import load_dotenv
from supabase import create_client

from experiment_context import get_experiment_context, with_experiment_context

load_dotenv()

PAPER_SYMBOL = "tTESTBTC:TESTUSD"
SIGNAL_SOURCE = "live_signal_generator_v1"
SIGNAL_REASON = "latest_mts_newer_than_last_generated_signal_mts"
SIGNAL_CONFIDENCE = 0.55

SUPABASE_URL = os.getenv("SUPABASE_URL")
SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
SYMBOL = os.getenv("SYMBOL", PAPER_SYMBOL)
TIMEFRAME = os.getenv("TIMEFRAME", "1m")


def parse_positive_int_env(name, default):
    raw_value = os.getenv(name)

    if raw_value in (None, ""):
        return default

    try:
        value = int(raw_value)
    except ValueError as exc:
        raise RuntimeError(f"{name} must be an integer.") from exc

    if value <= 0:
        raise RuntimeError(f"{name} must be greater than zero.")

    return value


SIGNAL_POLL_SECONDS = parse_positive_int_env("SIGNAL_POLL_SECONDS", 30)


def fail_closed_check():
    if not SUPABASE_URL or not SUPABASE_SERVICE_ROLE_KEY:
        raise RuntimeError("Missing required Supabase env values.")

    if SYMBOL != PAPER_SYMBOL:
        raise RuntimeError(f"SYMBOL must be {PAPER_SYMBOL} for paper trading.")

    if not TIMEFRAME:
        raise RuntimeError("TIMEFRAME is required.")


def audit(supabase, event_type, severity="info", signal_id=None, payload=None):
    supabase.table("agent_audit_log").insert({
        "event_type": event_type,
        "severity": severity,
        "signal_id": signal_id,
        "payload": with_experiment_context(payload),
    }).execute()


def fetch_latest_two_candles(supabase):
    result = (
        supabase.table("market_candles")
        .select("*")
        .eq("symbol", SYMBOL)
        .eq("timeframe", TIMEFRAME)
        .order("mts", desc=True)
        .limit(2)
        .execute()
    )

    return result.data or []


def has_open_signal(supabase):
    result = (
        supabase.table("agent_signals")
        .select("id,status")
        .eq("symbol", SYMBOL)
        .or_("status.eq.queued,status.eq.processing")
        .limit(1)
        .execute()
    )

    return bool(result.data)


def fetch_last_generated_signal_mts(supabase):
    result = (
        supabase.table("agent_signals")
        .select("metadata")
        .eq("symbol", SYMBOL)
        .eq("source", SIGNAL_SOURCE)
        .order("created_at", desc=True)
        .limit(1)
        .execute()
    )

    if not result.data:
        return None

    metadata = result.data[0].get("metadata") or {}
    return metadata.get("latest_mts")


def insert_signal(supabase, latest, previous):
    metadata = {
        "experiment": get_experiment_context(),
        "latest_mts": latest["mts"],
        "previous_mts": previous["mts"],
        "latest_close": latest["close"],
        "previous_close": previous["close"],
    }

    result = supabase.table("agent_signals").insert({
        "symbol": SYMBOL,
        "side": "buy",
        "confidence": SIGNAL_CONFIDENCE,
        "source": SIGNAL_SOURCE,
        "reason": SIGNAL_REASON,
        "status": "queued",
        "metadata": metadata,
    }).execute()

    return result.data[0], metadata


def main():
    fail_closed_check()
    supabase = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)

    print("Signal Generator started.")
    print(
        f"Symbol: {SYMBOL} | timeframe: {TIMEFRAME} | "
        f"poll_seconds: {SIGNAL_POLL_SECONDS}"
    )

    audit(
        supabase,
        "signal_generator_started",
        payload={
            "symbol": SYMBOL,
            "timeframe": TIMEFRAME,
            "signal_poll_seconds": SIGNAL_POLL_SECONDS,
        },
    )

    while True:
        try:
            candles = fetch_latest_two_candles(supabase)

            if len(candles) < 2:
                print("Skipped signal: waiting for at least 2 candles.")
                time.sleep(SIGNAL_POLL_SECONDS)
                continue

            latest = candles[0]
            previous = candles[1]

            if has_open_signal(supabase):
                print("Skipped signal: queued or processing signal already exists.")
                time.sleep(SIGNAL_POLL_SECONDS)
                continue

            last_generated_signal_mts = fetch_last_generated_signal_mts(supabase)

            if last_generated_signal_mts is not None and latest["mts"] <= last_generated_signal_mts:
                print("Skipped signal: latest candle is not newer than last generated signal.")
                time.sleep(SIGNAL_POLL_SECONDS)
                continue

            signal, metadata = insert_signal(supabase, latest, previous)

            audit(
                supabase,
                "generated_signal",
                signal_id=signal["id"],
                payload={
                    "symbol": SYMBOL,
                    "side": "buy",
                    "confidence": SIGNAL_CONFIDENCE,
                    "metadata": metadata,
                },
            )

            print(
                f"Generated signal: id={signal['id']} symbol={SYMBOL} side=buy "
                f"latest_close={latest['close']} previous_close={previous['close']}"
            )
        except Exception as exc:
            print(f"ERROR: {exc}")
            try:
                audit(
                    supabase,
                    "signal_generator_error",
                    severity="error",
                    payload={"error": str(exc)},
                )
            except Exception as audit_exc:
                print(f"ERROR writing signal_generator_error audit event: {audit_exc}")

        time.sleep(SIGNAL_POLL_SECONDS)


if __name__ == "__main__":
    main()