ai_market_watcher.py

import json
import os
import time
from datetime import datetime, timezone

from dotenv import load_dotenv
from supabase import create_client

import openai_client
from experiment_context import get_experiment_context, with_experiment_context

load_dotenv()

PAPER_SYMBOL = "tTESTBTC:TESTUSD"
DEFAULT_AI_REQUEST_TIMEOUT_SECONDS = 20
DEFAULT_AI_REQUEST_MAX_ATTEMPTS = 2
DEFAULT_AI_REQUEST_BACKOFF_SECONDS = 2
DEFAULT_AI_DECISION_MAX_OUTPUT_TOKENS = 900
DEFAULT_AI_REFLECTION_MAX_OUTPUT_TOKENS = 700
ALLOWED_DECISIONS = {"BUY", "SELL", "HOLD"}
ALLOWED_CONFIDENCE_ADJUSTMENTS = {"none", "reduce", "increase"}
ALLOWED_REFLECTION_CONFIDENCE_REASSESSMENTS = {"too_low", "about_right", "too_high", "unknown"}
ALLOWED_AI_API_MODES = openai_client.SUPPORTED_API_MODES

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")
AI_API_MODE = os.getenv("AI_API_MODE", openai_client.API_MODE_CHAT_COMPLETIONS).strip().lower()
AI_API_URL = os.getenv("AI_API_URL", openai_client.DEFAULT_CHAT_COMPLETIONS_API_URL)
AI_RESPONSES_API_URL = os.getenv(
    "AI_RESPONSES_API_URL",
    openai_client.DEFAULT_RESPONSES_API_URL,
)
AI_API_KEY = os.getenv("AI_API_KEY") or os.getenv("OPENAI_API_KEY")
AI_MODEL = os.getenv("AI_MODEL", "gpt-4o-mini")
AI_DECISION_MODEL = os.getenv("AI_DECISION_MODEL") or AI_MODEL
AI_REFLECTION_MODEL = os.getenv("AI_REFLECTION_MODEL") or AI_MODEL
ORDERBOOK_TABLE = os.getenv("ORDERBOOK_TABLE", "orderbook_snapshots")
ORDERBOOK_TIME_COLUMN = os.getenv("ORDERBOOK_TIME_COLUMN", "created_at")
LIQUIDITY_TABLE = os.getenv("LIQUIDITY_TABLE", "liquidity_metrics")
LIQUIDITY_TIME_COLUMN = os.getenv("LIQUIDITY_TIME_COLUMN", "created_at")


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


def parse_non_negative_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 zero or greater.")

    return value


AI_POLL_SECONDS = parse_positive_int_env("AI_POLL_SECONDS", 30)
AI_REVIEW_DELAY_SECONDS = parse_positive_int_env("AI_REVIEW_DELAY_SECONDS", 120)
AI_REVIEW_SCAN_LIMIT = parse_positive_int_env("AI_REVIEW_SCAN_LIMIT", 50)
AI_REFLECTION_SCAN_LIMIT = parse_positive_int_env("AI_REFLECTION_SCAN_LIMIT", 20)
AI_REFLECTION_MAX_PER_CYCLE = parse_non_negative_int_env("AI_REFLECTION_MAX_PER_CYCLE", 1)
CANDLE_LOOKBACK = parse_positive_int_env("AI_CANDLE_LOOKBACK", 10)
AI_REQUEST_TIMEOUT_SECONDS = parse_positive_int_env(
    "AI_REQUEST_TIMEOUT_SECONDS",
    DEFAULT_AI_REQUEST_TIMEOUT_SECONDS,
)
AI_REQUEST_MAX_ATTEMPTS = parse_positive_int_env(
    "AI_REQUEST_MAX_ATTEMPTS",
    DEFAULT_AI_REQUEST_MAX_ATTEMPTS,
)
AI_REQUEST_BACKOFF_SECONDS = parse_positive_int_env(
    "AI_REQUEST_BACKOFF_SECONDS",
    DEFAULT_AI_REQUEST_BACKOFF_SECONDS,
)
AI_DECISION_MAX_OUTPUT_TOKENS = parse_positive_int_env(
    "AI_DECISION_MAX_OUTPUT_TOKENS",
    DEFAULT_AI_DECISION_MAX_OUTPUT_TOKENS,
)
AI_REFLECTION_MAX_OUTPUT_TOKENS = parse_positive_int_env(
    "AI_REFLECTION_MAX_OUTPUT_TOKENS",
    DEFAULT_AI_REFLECTION_MAX_OUTPUT_TOKENS,
)


def fail_closed_check():
    missing = []

    for name, value in {
        "SUPABASE_URL": SUPABASE_URL,
        "SUPABASE_SERVICE_ROLE_KEY": SUPABASE_SERVICE_ROLE_KEY,
        "AI_API_KEY or OPENAI_API_KEY": AI_API_KEY,
    }.items():
        if not value:
            missing.append(name)

    if missing:
        raise RuntimeError(f"Missing required env values: {', '.join(missing)}")

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

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

    if AI_API_MODE not in ALLOWED_AI_API_MODES:
        allowed_modes = ", ".join(sorted(ALLOWED_AI_API_MODES))
        raise RuntimeError(f"AI_API_MODE must be one of: {allowed_modes}.")


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_recent_candles(supabase, limit=CANDLE_LOOKBACK):
    result = (
        supabase.table("market_candles")
        .select("*")
        .eq("symbol", SYMBOL)
        .eq("timeframe", TIMEFRAME)
        .order("mts", desc=True)
        .limit(limit)
        .execute()
    )

    rows = result.data or []
    return list(reversed(rows))


def fetch_latest_row(supabase, table_name, time_column):
    result = (
        supabase.table(table_name)
        .select("*")
        .eq("symbol", SYMBOL)
        .order(time_column, desc=True)
        .limit(1)
        .execute()
    )

    if not result.data:
        return None

    return result.data[0]


def fetch_last_decision_event(supabase):
    result = (
        supabase.table("agent_audit_log")
        .select("created_at,payload")
        .eq("event_type", "ai_market_decision")
        .order("created_at", desc=True)
        .limit(1)
        .execute()
    )

    if not result.data:
        return None

    return result.data[0]


def fetch_recent_decision_events(supabase, limit=AI_REVIEW_SCAN_LIMIT):
    result = (
        supabase.table("agent_audit_log")
        .select("created_at,payload")
        .eq("event_type", "ai_market_decision")
        .order("created_at", desc=True)
        .limit(limit)
        .execute()
    )

    return result.data or []


def fetch_recent_review_events(supabase, limit=AI_REFLECTION_SCAN_LIMIT):
    result = (
        supabase.table("agent_audit_log")
        .select("created_at,payload")
        .eq("event_type", "ai_market_post_decision_review")
        .order("created_at", desc=True)
        .limit(limit)
        .execute()
    )

    return result.data or []


def fetch_event_by_decision_key(supabase, event_type, decision_key):
    result = (
        supabase.table("agent_audit_log")
        .select("created_at,payload")
        .eq("event_type", event_type)
        .contains("payload", {"decision_key": decision_key})
        .order("created_at", desc=True)
        .limit(1)
        .execute()
    )

    if not result.data:
        return None

    return result.data[0]


def build_decision_key(latest_mts):
    return f"{SYMBOL}:{TIMEFRAME}:{latest_mts}"


def decision_exists(supabase, decision_key):
    result = (
        supabase.table("agent_audit_log")
        .select("created_at")
        .eq("event_type", "ai_market_decision")
        .contains("payload", {"decision_key": decision_key})
        .limit(1)
        .execute()
    )

    return bool(result.data)


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 ai_signal_exists(supabase, decision_key):
    result = (
        supabase.table("agent_signals")
        .select("id,status")
        .eq("symbol", SYMBOL)
        .eq("source", "ai_market_watcher_v1")
        .contains("metadata", {"decision_key": decision_key})
        .limit(1)
        .execute()
    )

    return bool(result.data)


def review_exists(supabase, decision_key):
    result = (
        supabase.table("agent_audit_log")
        .select("created_at")
        .eq("event_type", "ai_market_post_decision_review")
        .contains("payload", {"decision_key": decision_key})
        .limit(1)
        .execute()
    )

    return bool(result.data)


def reflection_exists(supabase, decision_key):
    result = (
        supabase.table("agent_audit_log")
        .select("created_at")
        .eq("event_type", "ai_market_reflection")
        .contains("payload", {"decision_key": decision_key})
        .limit(1)
        .execute()
    )

    return bool(result.data)


def parse_created_at(value):
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


def build_decision_policy():
    return {
        "decision_contract": (
            "BUY and SELL mean queue-worthy short-term directional intent for the paper agent. "
            "HOLD is the correct active decision when the setup is mixed, trap-like, or mainly says to avoid entry."
        ),
        "rules": [
            "Use BUY only when upward momentum is supported by buy pressure, improving or stable liquidity, and low or medium bad-entry risk.",
            "Use SELL only when downward momentum or a clear failed bounce is supported by sell pressure and enough evidence for downside continuation.",
            "Do not use SELL merely because ask-side resistance is heavy, the spread is wide, or a long entry looks bad.",
            "Use HOLD for liquidity-trap patterns: recent price pop into heavy ask-side pressure, widening spread, weakening volume, or high bad-entry risk.",
            "When momentum and liquidity/orderbook evidence disagree, prefer HOLD unless the disagreement clearly confirms a breakdown.",
            "Confidence should reflect decision quality. A high-confidence HOLD is valid when the best action is to avoid a bad entry.",
        ],
        "decision_modes": [
            "momentum_buy",
            "breakdown_sell",
            "liquidity_trap_hold",
            "neutral_hold",
            "uncertain_hold",
        ],
        "missing_context_policy": (
            "If position, fees, slippage, or prior signal context would be required to justify BUY or SELL, "
            "prefer HOLD and name the missing context in the premortem."
        ),
    }


def build_decision_response_schema():
    return {
        "type": "object",
        "additionalProperties": False,
        "properties": {
            "decision": {"type": "string", "enum": ["BUY", "SELL", "HOLD"]},
            "confidence": {"type": "number", "minimum": 0, "maximum": 1},
            "reason": {"type": "string"},
            "decision_mode": {
                "type": "string",
                "enum": [
                    "momentum_buy",
                    "breakdown_sell",
                    "liquidity_trap_hold",
                    "neutral_hold",
                    "uncertain_hold",
                ],
            },
            "market_summary": {
                "type": "object",
                "additionalProperties": False,
                "properties": {
                    "momentum": {"type": "string", "enum": ["up", "down", "flat"]},
                    "directional_pressure": {
                        "type": "string",
                        "enum": ["buy", "sell", "neutral"],
                    },
                    "bad_entry_risk": {
                        "type": "string",
                        "enum": ["low", "medium", "high"],
                    },
                    "liquidity_shift": {
                        "type": "string",
                        "enum": ["improving", "weakening", "stable"],
                    },
                },
                "required": [
                    "momentum",
                    "directional_pressure",
                    "bad_entry_risk",
                    "liquidity_shift",
                ],
            },
            "shift_summary": {
                "type": "object",
                "additionalProperties": False,
                "properties": {
                    "momentum_shift": {
                        "type": "string",
                        "enum": [
                            "flat_to_up",
                            "flat_to_down",
                            "up_to_down",
                            "down_to_up",
                            "no_change",
                            "unknown",
                        ],
                    },
                    "liquidity_shift": {
                        "type": "string",
                        "enum": [
                            "stable_to_weakening",
                            "stable_to_improving",
                            "weakening_to_improving",
                            "improving_to_weakening",
                            "no_change",
                            "unknown",
                        ],
                    },
                    "pressure_shift": {
                        "type": "string",
                        "enum": [
                            "neutral_to_buy",
                            "neutral_to_sell",
                            "buy_to_sell",
                            "sell_to_buy",
                            "no_change",
                            "unknown",
                        ],
                    },
                },
                "required": ["momentum_shift", "liquidity_shift", "pressure_shift"],
            },
            "premortem": {
                "type": "object",
                "additionalProperties": False,
                "properties": {
                    "failure_scenario": {"type": "string"},
                    "most_likely_cause": {"type": "string"},
                    "what_i_might_be_missing": {
                        "type": "array",
                        "items": {"type": "string"},
                    },
                    "warning_signs": {
                        "type": "array",
                        "items": {"type": "string"},
                    },
                    "confidence_adjustment": {
                        "type": "string",
                        "enum": ["none", "reduce", "increase"],
                    },
                    "final_reflection": {"type": "string"},
                },
                "required": [
                    "failure_scenario",
                    "most_likely_cause",
                    "what_i_might_be_missing",
                    "warning_signs",
                    "confidence_adjustment",
                    "final_reflection",
                ],
            },
            "risks": {
                "type": "array",
                "items": {"type": "string"},
            },
            "experimental_notes": {
                "type": "array",
                "items": {"type": "string"},
            },
        },
        "required": [
            "decision",
            "confidence",
            "reason",
            "decision_mode",
            "market_summary",
            "shift_summary",
            "premortem",
            "risks",
            "experimental_notes",
        ],
    }


def build_reflection_response_schema():
    return {
        "type": "object",
        "additionalProperties": False,
        "properties": {
            "what_was_right": {"type": "string"},
            "what_was_wrong": {"type": "string"},
            "missed_warning_signs": {
                "type": "array",
                "items": {"type": "string"},
            },
            "confidence_reassessment": {
                "type": "string",
                "enum": ["too_low", "about_right", "too_high", "unknown"],
            },
            "should_have_decided": {
                "type": "string",
                "enum": ["BUY", "SELL", "HOLD", "UNKNOWN"],
            },
            "lesson_for_next_decision": {"type": "string"},
            "next_observation": {"type": "string"},
            "reflection_note": {"type": "string"},
            "experimental_notes": {
                "type": "array",
                "items": {"type": "string"},
            },
        },
        "required": [
            "what_was_right",
            "what_was_wrong",
            "missed_warning_signs",
            "confidence_reassessment",
            "should_have_decided",
            "lesson_for_next_decision",
            "next_observation",
            "reflection_note",
            "experimental_notes",
        ],
    }


def build_decision_system_prompt():
    static_context = {
        "role": "lightweight_market_watcher",
        "output_contract": "Return only the structured response requested by the API schema.",
        "scope": "Market awareness testing for a paper-only prototype, not profitability.",
        "decision_policy": build_decision_policy(),
        "task": {
            "goal": "Assess short-term market awareness.",
            "focus": [
                "momentum_detection",
                "directional_pressure",
                "bad_entry_avoidance",
                "liquidity_shift_reaction",
            ],
        },
        "shift_detection_task": {
            "goal": "Compare current market context with the previous AI market summary and detect state transitions.",
            "detect_transitions": [
                "flat_to_up",
                "flat_to_down",
                "stable_to_weakening",
                "stable_to_improving",
                "neutral_to_buy_pressure",
                "neutral_to_sell_pressure",
            ],
        },
        "experimental_notes_policy": (
            "Use experimental_notes for useful observations that are not part of the stable decision contract. "
            "These notes are audit-only and must not be required to execute a trade."
        ),
    }
    return json.dumps(static_context, separators=(",", ":"), sort_keys=True)


def build_reflection_system_prompt():
    static_context = {
        "role": "market_awareness_reflector",
        "output_contract": "Return only the structured response requested by the API schema.",
        "scope": "Audit-only reflection. Do not propose or execute a trade.",
        "task": {
            "goal": "Reflect on decision quality after observing the later market move.",
            "constraints": [
                "Do not optimize for profit or propose a live trade.",
                "Do not rewrite the original decision.",
                "Identify what the original reasoning missed or handled well.",
                "Produce a small lesson for the next market-awareness decision.",
            ],
        },
        "experimental_notes_policy": (
            "Use experimental_notes for useful observations that are not part of the stable reflection contract."
        ),
    }
    return json.dumps(static_context, separators=(",", ":"), sort_keys=True)


def build_payload(supabase, candles, orderbook, liquidity):
    latest = candles[-1]
    previous = candles[-2] if len(candles) >= 2 else None
    last_decision = fetch_last_decision_event(supabase)
    previous_state = (last_decision or {}).get("payload", {}).get("market_summary", {})

    recent_closes = [float(candle["close"]) for candle in candles[-5:]]
    recent_volumes = [float(candle["volume"]) for candle in candles[-5:]]

    return {
        "symbol": SYMBOL,
        "timeframe": TIMEFRAME,
        "latest_candle": latest,
        "previous_candle": previous,
        "recent_closes": recent_closes,
        "recent_volumes": recent_volumes,
        "orderbook_snapshot": orderbook,
        "liquidity_metrics": liquidity,
        "previous_market_summary": previous_state,
    }


def normalize_premortem(raw_premortem):
    if raw_premortem is None:
        return None, False

    if not isinstance(raw_premortem, dict):
        return None, False

    is_valid = True

    raw_missing_items = raw_premortem.get("what_i_might_be_missing") or []
    if not isinstance(raw_missing_items, list):
        raw_missing_items = []
        is_valid = False

    raw_warning_signs = raw_premortem.get("warning_signs") or []
    if not isinstance(raw_warning_signs, list):
        raw_warning_signs = []
        is_valid = False

    confidence_adjustment = str(raw_premortem.get("confidence_adjustment", "none")).strip().lower()
    if confidence_adjustment not in ALLOWED_CONFIDENCE_ADJUSTMENTS:
        confidence_adjustment = "none"
        is_valid = False

    premortem = {
        "failure_scenario": str(raw_premortem.get("failure_scenario", "")).strip(),
        "most_likely_cause": str(raw_premortem.get("most_likely_cause", "")).strip(),
        "what_i_might_be_missing": [str(item) for item in raw_missing_items],
        "warning_signs": [str(item) for item in raw_warning_signs],
        "confidence_adjustment": confidence_adjustment,
        "final_reflection": str(raw_premortem.get("final_reflection", "")).strip(),
    }

    if not premortem["failure_scenario"]:
        is_valid = False
    if not premortem["most_likely_cause"]:
        is_valid = False
    if not premortem["final_reflection"]:
        is_valid = False

    return premortem, is_valid


def log_ai_request_retry(supabase, event):
    if not supabase:
        return

    audit(
        supabase,
        "ai_request_retry",
        severity="warning",
        payload=event,
    )


def make_ai_request_retry_logger(supabase):
    def on_retry(event):
        log_ai_request_retry(supabase, event)
        status = event.get("status_code")
        if status:
            print(
                f"WARNING: OpenAI {event['request_kind']} returned HTTP {status}; "
                f"retrying in {event['backoff_seconds']}s."
            )
        else:
            print(
                f"WARNING: OpenAI {event['request_kind']} request failed with "
                f"{event['error_type']}; retrying in {event['backoff_seconds']}s."
            )

    return on_retry


def build_openai_request_metadata(result, model):
    return {
        "api_mode": AI_API_MODE,
        "model": model,
        "status_code": result.get("status_code"),
        "request_id": result.get("request_id"),
        "latency_ms": result.get("latency_ms"),
        "attempts": result.get("attempts"),
        "usage": result.get("usage"),
    }


def ask_ai(payload, supabase=None):
    result = openai_client.request_structured_json(
        api_mode=AI_API_MODE,
        chat_completions_api_url=AI_API_URL,
        responses_api_url=AI_RESPONSES_API_URL,
        api_key=AI_API_KEY,
        model=AI_DECISION_MODEL,
        system_prompt=build_decision_system_prompt(),
        user_payload=payload,
        schema_name="ai_market_decision",
        schema=build_decision_response_schema(),
        request_kind="decision",
        timeout_seconds=AI_REQUEST_TIMEOUT_SECONDS,
        max_attempts=AI_REQUEST_MAX_ATTEMPTS,
        backoff_seconds=AI_REQUEST_BACKOFF_SECONDS,
        max_output_tokens=AI_DECISION_MAX_OUTPUT_TOKENS,
        on_retry=make_ai_request_retry_logger(supabase),
    )
    decision = validate_ai_decision(result["data"])
    decision["openai_request"] = build_openai_request_metadata(result, AI_DECISION_MODEL)
    return decision


def ask_ai_reflection(payload, supabase=None):
    result = openai_client.request_structured_json(
        api_mode=AI_API_MODE,
        chat_completions_api_url=AI_API_URL,
        responses_api_url=AI_RESPONSES_API_URL,
        api_key=AI_API_KEY,
        model=AI_REFLECTION_MODEL,
        system_prompt=build_reflection_system_prompt(),
        user_payload=payload,
        schema_name="ai_market_reflection",
        schema=build_reflection_response_schema(),
        request_kind="reflection",
        timeout_seconds=AI_REQUEST_TIMEOUT_SECONDS,
        max_attempts=AI_REQUEST_MAX_ATTEMPTS,
        backoff_seconds=AI_REQUEST_BACKOFF_SECONDS,
        max_output_tokens=AI_REFLECTION_MAX_OUTPUT_TOKENS,
        on_retry=make_ai_request_retry_logger(supabase),
    )
    reflection = validate_ai_reflection(result["data"])
    reflection["openai_request"] = build_openai_request_metadata(result, AI_REFLECTION_MODEL)
    return reflection


def validate_ai_decision(decision):
    if not isinstance(decision, dict):
        raise RuntimeError("AI decision must be a JSON object.")

    normalized_decision = str(decision.get("decision", "")).upper()
    if normalized_decision not in ALLOWED_DECISIONS:
        raise RuntimeError("AI decision must be BUY, SELL, or HOLD.")

    try:
        confidence = float(decision.get("confidence", 0))
    except (TypeError, ValueError) as exc:
        raise RuntimeError("AI confidence must be numeric.") from exc

    confidence = max(0.0, min(confidence, 1.0))
    reason = str(decision.get("reason", "")).strip()
    if not reason:
        raise RuntimeError("AI reason is required.")

    market_summary = decision.get("market_summary") or {}
    if not isinstance(market_summary, dict):
        raise RuntimeError("AI market_summary must be a JSON object.")

    shift_summary = decision.get("shift_summary") or {}
    if not isinstance(shift_summary, dict):
        raise RuntimeError("AI shift_summary must be a JSON object.")

    decision_mode = decision.get("decision_mode")
    if decision_mode is not None:
        decision_mode = str(decision_mode).strip() or None

    premortem, premortem_is_valid = normalize_premortem(decision.get("premortem"))

    risks = decision.get("risks") or []
    if not isinstance(risks, list):
        raise RuntimeError("AI risks must be a list.")

    experimental_notes = decision.get("experimental_notes") or []
    if not isinstance(experimental_notes, list):
        raise RuntimeError("AI experimental_notes must be a list.")

    return {
        "decision": normalized_decision,
        "confidence": confidence,
        "reason": reason,
        "decision_mode": decision_mode,
        "market_summary": market_summary,
        "shift_summary": shift_summary,
        "premortem": premortem,
        "premortem_is_valid": premortem_is_valid,
        "risks": [str(item) for item in risks],
        "experimental_notes": [str(item) for item in experimental_notes],
    }


def validate_ai_reflection(reflection):
    if not isinstance(reflection, dict):
        raise RuntimeError("AI reflection must be a JSON object.")

    is_valid = True

    def normalize_text(name):
        nonlocal is_valid
        value = str(reflection.get(name, "")).strip()
        if not value:
            is_valid = False
        return value

    missed_warning_signs = reflection.get("missed_warning_signs") or []
    if not isinstance(missed_warning_signs, list):
        missed_warning_signs = []
        is_valid = False

    experimental_notes = reflection.get("experimental_notes") or []
    if not isinstance(experimental_notes, list):
        experimental_notes = []
        is_valid = False

    confidence_reassessment = (
        str(reflection.get("confidence_reassessment", "unknown")).strip().lower()
    )
    if confidence_reassessment not in ALLOWED_REFLECTION_CONFIDENCE_REASSESSMENTS:
        confidence_reassessment = "unknown"
        is_valid = False

    should_have_decided = str(reflection.get("should_have_decided", "UNKNOWN")).strip().upper()
    if should_have_decided not in ALLOWED_DECISIONS and should_have_decided != "UNKNOWN":
        should_have_decided = "UNKNOWN"
        is_valid = False

    return {
        "what_was_right": normalize_text("what_was_right"),
        "what_was_wrong": normalize_text("what_was_wrong"),
        "missed_warning_signs": [str(item) for item in missed_warning_signs],
        "confidence_reassessment": confidence_reassessment,
        "should_have_decided": should_have_decided,
        "lesson_for_next_decision": normalize_text("lesson_for_next_decision"),
        "next_observation": normalize_text("next_observation"),
        "reflection_note": normalize_text("reflection_note"),
        "experimental_notes": [str(item) for item in experimental_notes],
        "reflection_is_valid": is_valid,
    }


def log_decision(supabase, candles, orderbook, liquidity, ai_decision):
    latest = candles[-1]
    previous = candles[-2] if len(candles) >= 2 else None
    latest_close = float(latest["close"])
    previous_close = float(previous["close"]) if previous else None
    decision_key = build_decision_key(latest["mts"])

    payload = {
        "decision_key": decision_key,
        "symbol": SYMBOL,
        "timeframe": TIMEFRAME,
        "experiment": get_experiment_context(),
        "latest_mts": latest["mts"],
        "latest_close": latest_close,
        "previous_mts": previous["mts"] if previous else None,
        "previous_close": previous_close,
        "decision": ai_decision["decision"],
        "confidence": ai_decision["confidence"],
        "reason": ai_decision["reason"],
        "decision_mode": ai_decision.get("decision_mode"),
        "market_summary": ai_decision["market_summary"],
        "shift_summary": ai_decision["shift_summary"],
        "risks": ai_decision["risks"],
        "experimental_notes": ai_decision.get("experimental_notes") or [],
        "openai_request": ai_decision.get("openai_request"),
        "orderbook_table": ORDERBOOK_TABLE,
        "orderbook_time": orderbook.get(ORDERBOOK_TIME_COLUMN) if orderbook else None,
        "liquidity_table": LIQUIDITY_TABLE,
        "liquidity_time": liquidity.get(LIQUIDITY_TIME_COLUMN) if liquidity else None,
    }

    audit(supabase, "ai_market_decision", payload=payload)
    return decision_key


def log_shift_detected(supabase, decision_key, previous_market_summary, ai_decision):
    shift_summary = ai_decision.get("shift_summary") or {}

    if not any(value != "no_change" for value in shift_summary.values()):
        return

    payload = {
        "symbol": SYMBOL,
        "timeframe": TIMEFRAME,
        "decision_key": decision_key,
        "previous_market_summary": previous_market_summary or {},
        "current_market_summary": ai_decision.get("market_summary") or {},
        "shift_summary": shift_summary,
        "decision": ai_decision.get("decision"),
        "confidence": ai_decision.get("confidence"),
        "reason": ai_decision.get("reason"),
    }

    audit(
        supabase,
        "ai_market_shift_detected",
        severity="info",
        payload=payload,
    )


def log_decision_premortem(supabase, decision_key, ai_decision):
    premortem = ai_decision.get("premortem")

    if not premortem or not ai_decision.get("premortem_is_valid"):
        audit(
            supabase,
            "ai_decision_premortem_missing",
            severity="info",
            payload={
                "symbol": SYMBOL,
                "timeframe": TIMEFRAME,
                "decision_key": decision_key,
                "decision": ai_decision.get("decision"),
                "confidence": ai_decision.get("confidence"),
                "reason": ai_decision.get("reason"),
                "decision_mode": ai_decision.get("decision_mode"),
                "experimental_notes": ai_decision.get("experimental_notes") or [],
                "openai_request": ai_decision.get("openai_request"),
            },
        )
        return

    payload = {
        "symbol": SYMBOL,
        "timeframe": TIMEFRAME,
        "decision_key": decision_key,
        "decision": ai_decision.get("decision"),
        "confidence": ai_decision.get("confidence"),
        "reason": ai_decision.get("reason"),
        "market_summary": ai_decision.get("market_summary") or {},
        "shift_summary": ai_decision.get("shift_summary") or {},
        "premortem": premortem,
        "decision_mode": ai_decision.get("decision_mode"),
        "experimental_notes": ai_decision.get("experimental_notes") or [],
        "openai_request": ai_decision.get("openai_request"),
    }

    audit(
        supabase,
        "ai_decision_premortem",
        severity="info",
        payload=payload,
    )


def log_trade_premortem(supabase, decision_key, ai_decision):
    if ai_decision.get("decision") not in {"BUY", "SELL"}:
        return

    premortem = ai_decision.get("premortem")
    if not isinstance(premortem, dict):
        return

    payload = {
        "symbol": SYMBOL,
        "timeframe": TIMEFRAME,
        "decision_key": decision_key,
        "decision": ai_decision.get("decision"),
        "confidence": ai_decision.get("confidence"),
        "reason": ai_decision.get("reason"),
        "decision_mode": ai_decision.get("decision_mode"),
        "market_summary": ai_decision.get("market_summary") or {},
        "shift_summary": ai_decision.get("shift_summary") or {},
        "failure_scenario": premortem.get("failure_scenario", ""),
        "most_likely_cause": premortem.get("most_likely_cause", ""),
        "warning_signs": premortem.get("warning_signs", []),
        "what_i_might_be_missing": premortem.get("what_i_might_be_missing", []),
        "confidence_adjustment": premortem.get("confidence_adjustment", "none"),
        "final_reflection": premortem.get("final_reflection", ""),
        "experimental_notes": ai_decision.get("experimental_notes") or [],
        "openai_request": ai_decision.get("openai_request"),
    }

    audit(
        supabase,
        "ai_trade_premortem",
        severity="info",
        payload=payload,
    )


def insert_ai_signal(supabase, candles, orderbook, liquidity, ai_decision, decision_key):
    latest = candles[-1]

    side = ai_decision["decision"].lower()
    if side not in {"buy", "sell"}:
        return None

    metadata = {
        "source": "ai_market_watcher",
        "experiment": get_experiment_context(),
        "decision_key": decision_key,
        "latest_mts": latest["mts"],
        "latest_close": latest["close"],
        "timeframe": TIMEFRAME,
        "ai_confidence": ai_decision["confidence"],
        "ai_reason": ai_decision["reason"],
        "market_summary": ai_decision["market_summary"],
        "shift_summary": ai_decision["shift_summary"],
        "risks": ai_decision["risks"],
        "experimental_notes": ai_decision.get("experimental_notes") or [],
        "openai_request": ai_decision.get("openai_request"),
        "orderbook_snapshot_id": orderbook.get("id"),
        "liquidity_metrics_id": liquidity.get("id"),
    }

    result = supabase.table("agent_signals").insert({
        "symbol": SYMBOL,
        "side": side,
        "confidence": ai_decision["confidence"],
        "source": "ai_market_watcher_v1",
        "reason": ai_decision["reason"],
        "status": "queued",
        "metadata": metadata,
    }).execute()

    return result.data[0]


def build_post_decision_review_payload(decision_event, latest_candle):
    payload = decision_event.get("payload") or {}
    decision_key = payload.get("decision_key")
    decision_created_at = decision_event.get("created_at")
    decision_latest_mts = payload.get("latest_mts")
    decision_latest_close = payload.get("latest_close")

    if not decision_key or decision_latest_mts is None or decision_latest_close is None:
        return None

    created_at = parse_created_at(decision_created_at)
    age_seconds = (datetime.now(timezone.utc) - created_at).total_seconds()

    if latest_candle["mts"] <= decision_latest_mts:
        return None

    if age_seconds < AI_REVIEW_DELAY_SECONDS:
        return None

    latest_close = float(latest_candle["close"])
    decision_close = float(decision_latest_close)
    price_change = latest_close - decision_close
    price_change_pct = 0.0

    if decision_close != 0:
        price_change_pct = (price_change / decision_close) * 100.0

    if price_change > 0:
        movement = "up"
    elif price_change < 0:
        movement = "down"
    else:
        movement = "flat"

    return {
        "decision_key": decision_key,
        "decision": payload.get("decision"),
        "decision_created_at": decision_created_at,
        "decision_latest_mts": decision_latest_mts,
        "review_latest_mts": latest_candle["mts"],
        "decision_close": decision_close,
        "review_close": latest_close,
        "price_change": price_change,
        "price_change_pct": price_change_pct,
        "movement": movement,
        "review_age_seconds": age_seconds,
        "review_mode": "latest_available_candle",
    }


def maybe_log_post_decision_reviews(supabase, candles):
    latest_candle = candles[-1]
    logged_count = 0

    for decision_event in reversed(fetch_recent_decision_events(supabase)):
        review_payload = build_post_decision_review_payload(decision_event, latest_candle)
        if not review_payload:
            continue

        decision_key = review_payload["decision_key"]
        if review_exists(supabase, decision_key):
            continue

        audit(supabase, "ai_market_post_decision_review", payload=review_payload)
        logged_count += 1
        print(
            f"Logged post-decision review: decision={review_payload.get('decision')} "
            f"movement={review_payload['movement']} "
            f"price_change_pct={review_payload['price_change_pct']:.4f} "
            f"decision_key={decision_key}"
        )

    return logged_count


def build_reflection_payload(decision_event, premortem_event, review_event):
    decision_payload = decision_event.get("payload") or {}
    premortem_payload = (premortem_event or {}).get("payload") or {}
    review_payload = review_event.get("payload") or {}

    return {
        "symbol": SYMBOL,
        "timeframe": TIMEFRAME,
        "decision_key": review_payload.get("decision_key"),
        "original_decision": {
            "created_at": decision_event.get("created_at"),
            "decision": decision_payload.get("decision"),
            "confidence": decision_payload.get("confidence"),
            "reason": decision_payload.get("reason"),
            "decision_mode": decision_payload.get("decision_mode"),
            "market_summary": decision_payload.get("market_summary") or {},
            "shift_summary": decision_payload.get("shift_summary") or {},
            "risks": decision_payload.get("risks") or [],
        },
        "original_premortem": premortem_payload.get("premortem"),
        "post_decision_review": {
            **review_payload,
            "created_at": review_event.get("created_at"),
        },
    }


def log_ai_reflection(supabase, reflection_payload, ai_reflection):
    decision_key = reflection_payload.get("decision_key")
    original_decision = reflection_payload.get("original_decision") or {}
    post_decision_review = reflection_payload.get("post_decision_review") or {}

    payload = {
        "symbol": SYMBOL,
        "timeframe": TIMEFRAME,
        "decision_key": decision_key,
        "ai_model": AI_MODEL,
        "original_decision": original_decision,
        "original_premortem": reflection_payload.get("original_premortem"),
        "post_decision_review": post_decision_review,
        "reflection": ai_reflection,
        "reflection_is_valid": ai_reflection.get("reflection_is_valid", False),
        "experimental_notes": ai_reflection.get("experimental_notes") or [],
        "openai_request": ai_reflection.get("openai_request"),
    }

    audit(supabase, "ai_market_reflection", severity="info", payload=payload)


def maybe_log_ai_reflections(supabase, max_count=AI_REFLECTION_MAX_PER_CYCLE):
    if max_count <= 0:
        return 0

    logged_count = 0

    for review_event in reversed(fetch_recent_review_events(supabase)):
        review_payload = review_event.get("payload") or {}
        decision_key = review_payload.get("decision_key")
        if not decision_key:
            continue

        if reflection_exists(supabase, decision_key):
            continue

        decision_event = fetch_event_by_decision_key(
            supabase,
            "ai_market_decision",
            decision_key,
        )
        if not decision_event:
            continue

        premortem_event = fetch_event_by_decision_key(
            supabase,
            "ai_decision_premortem",
            decision_key,
        )
        reflection_payload = build_reflection_payload(decision_event, premortem_event, review_event)
        ai_reflection = ask_ai_reflection(reflection_payload, supabase=supabase)

        if reflection_exists(supabase, decision_key):
            continue

        log_ai_reflection(supabase, reflection_payload, ai_reflection)
        logged_count += 1
        print(
            f"Logged AI reflection: decision_key={decision_key} "
            f"confidence_reassessment={ai_reflection['confidence_reassessment']} "
            f"should_have_decided={ai_reflection['should_have_decided']}"
        )

        if logged_count >= max_count:
            break

    return logged_count


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

    print("AI Market Watcher started.")
    print(
        f"Symbol: {SYMBOL} | timeframe: {TIMEFRAME} | poll_seconds: {AI_POLL_SECONDS} | "
        f"api_mode: {AI_API_MODE} | decision_model: {AI_DECISION_MODEL} | "
        f"reflection_model: {AI_REFLECTION_MODEL}"
    )

    audit(
        supabase,
        "ai_market_watcher_started",
        payload={
            "symbol": SYMBOL,
            "timeframe": TIMEFRAME,
            "ai_model": AI_MODEL,
            "ai_decision_model": AI_DECISION_MODEL,
            "ai_reflection_model": AI_REFLECTION_MODEL,
            "ai_api_mode": AI_API_MODE,
            "experiment": get_experiment_context(),
            "orderbook_table": ORDERBOOK_TABLE,
            "liquidity_table": LIQUIDITY_TABLE,
            "ai_poll_seconds": AI_POLL_SECONDS,
            "ai_review_delay_seconds": AI_REVIEW_DELAY_SECONDS,
            "ai_review_scan_limit": AI_REVIEW_SCAN_LIMIT,
            "ai_reflection_scan_limit": AI_REFLECTION_SCAN_LIMIT,
            "ai_reflection_max_per_cycle": AI_REFLECTION_MAX_PER_CYCLE,
            "ai_request_timeout_seconds": AI_REQUEST_TIMEOUT_SECONDS,
            "ai_request_max_attempts": AI_REQUEST_MAX_ATTEMPTS,
            "ai_request_backoff_seconds": AI_REQUEST_BACKOFF_SECONDS,
            "ai_decision_max_output_tokens": AI_DECISION_MAX_OUTPUT_TOKENS,
            "ai_reflection_max_output_tokens": AI_REFLECTION_MAX_OUTPUT_TOKENS,
        },
    )

    while True:
        try:
            candles = fetch_recent_candles(supabase)
            if len(candles) < 2:
                print("Skipped AI decision: waiting for at least 2 candles.")
                time.sleep(AI_POLL_SECONDS)
                continue

            orderbook = fetch_latest_row(supabase, ORDERBOOK_TABLE, ORDERBOOK_TIME_COLUMN)
            liquidity = fetch_latest_row(supabase, LIQUIDITY_TABLE, LIQUIDITY_TIME_COLUMN)

            if not orderbook or not liquidity:
                print("Skipped AI decision: missing orderbook snapshot or liquidity metrics.")
                time.sleep(AI_POLL_SECONDS)
                continue

            maybe_log_post_decision_reviews(supabase, candles)
            maybe_log_ai_reflections(supabase)

            latest_mts = candles[-1]["mts"]
            decision_key = build_decision_key(latest_mts)

            if decision_exists(supabase, decision_key):
                print("Skipped AI decision: decision already exists for latest candle.")
                time.sleep(AI_POLL_SECONDS)
                continue

            last_decision = fetch_last_decision_event(supabase)
            last_payload = (last_decision or {}).get("payload") or {}
            last_latest_mts = last_payload.get("latest_mts")

            if last_latest_mts is not None and latest_mts <= last_latest_mts:
                print("Skipped AI decision: latest candle already reviewed by AI.")
                time.sleep(AI_POLL_SECONDS)
                continue

            ai_payload = build_payload(supabase, candles, orderbook, liquidity)
            ai_decision = ask_ai(ai_payload, supabase=supabase)

            if decision_exists(supabase, decision_key):
                print("Skipped AI decision: decision was created during AI call.")
                time.sleep(AI_POLL_SECONDS)
                continue

            decision_key = log_decision(supabase, candles, orderbook, liquidity, ai_decision)
            log_decision_premortem(supabase, decision_key, ai_decision)
            log_shift_detected(
                supabase,
                decision_key,
                ai_payload.get("previous_market_summary") or {},
                ai_decision,
            )

            if ai_decision["decision"] in {"BUY", "SELL"}:
                log_trade_premortem(supabase, decision_key, ai_decision)

                if has_open_signal(supabase):
                    audit(
                        supabase,
                        "ai_signal_skipped_open_signal_exists",
                        payload={
                            "decision_key": decision_key,
                            "decision": ai_decision["decision"],
                        },
                    )
                    print("Skipped AI signal: queued or processing signal already exists.")
                elif ai_signal_exists(supabase, decision_key):
                    audit(
                        supabase,
                        "ai_signal_skipped_duplicate_decision_key",
                        payload={
                            "decision_key": decision_key,
                            "decision": ai_decision["decision"],
                        },
                    )
                    print("Skipped AI signal: signal already exists for decision_key.")
                else:
                    signal = insert_ai_signal(
                        supabase,
                        candles,
                        orderbook,
                        liquidity,
                        ai_decision,
                        decision_key,
                    )

                    audit(
                        supabase,
                        "ai_signal_queued",
                        signal_id=signal["id"],
                        payload={
                            "decision_key": decision_key,
                            "side": signal["side"],
                            "confidence": signal["confidence"],
                        },
                    )

                    print(f"AI signal queued: id={signal['id']} side={signal['side']}")
            else:
                print("AI decision was HOLD: audit only, no signal queued.")

            print(
                f"AI decision logged: decision={ai_decision['decision']} confidence={ai_decision['confidence']:.2f} "
                f"decision_key={decision_key}"
            )
        except Exception as exc:
            print(f"ERROR: {exc}")
            try:
                audit(
                    supabase,
                    "ai_market_watcher_error",
                    severity="error",
                    payload={"error": str(exc)},
                )
            except Exception as audit_exc:
                print(f"ERROR writing ai_market_watcher_error audit event: {audit_exc}")

        time.sleep(AI_POLL_SECONDS)


if __name__ == "__main__":
    main()