import { defineConnector, raw, type NormalizedBatch, type ProposedInstrument } from "@market-atlas/connector-sdk";
import type { NormalizedObservation, RawObservation } from "@market-atlas/market-model";
import { zonedTimeToUtc } from "@market-atlas/market-model";
import { US_EQUITIES, US_ETFS, US_INDICES, equityHint, indexHint } from "../_shared/us-universe.js";

const BASE = "https://cdn.cboe.com/api/global/delayed_quotes/quotes";
const ET = "America/New_York";

const seeds: ProposedInstrument[] = [
  ...US_EQUITIES.map(([s, n, v]) => ({ symbol: s, hint: equityHint(s, n, v, "EQUITY"), aliases: s.includes(".") ? [s.replace(".", "-"), s.replace(".", "/"), s.replace(".", " ")] : [] })),
  ...US_ETFS.map(([s, n, v]) => ({ symbol: s, hint: equityHint(s, n, v, "ETF") })),
  ...US_INDICES.map((i) => ({ symbol: i.cboe, hint: indexHint(i), aliases: [i.symbol, `^${i.symbol}`, `.${i.symbol}`] })),
];
const SYMBOLS = seeds.map((s) => s.symbol);

/**
 * Cboe public delayed quotes (JSON per symbol, the same endpoint the cboe.com quote pages use).
 * Adaptive polling: every 60 s per symbol during the US session, every 15 min when closed, hourly on
 * weekends. Requests are rate-limited centrally (3/s on cdn.cboe.com). Data is 15-minute delayed.
 */
export const cboeDelayedQuotes = defineConnector({
  metadata: {
    id: "cboe-delayed-quotes",
    name: "Cboe — delayed US quotes (equities, ETFs, indices)",
    version: "1.0.0",
    sourceId: "cboe",
    organization: "Cboe Global Markets",
    sourceType: "XHR",
    jurisdiction: "US",
    rightsStatus: "DELAYED",
    realtimeStatus: "DELAYED",
    expectedLatencyMs: 15 * 60_000,
    supportsStreaming: false,
    supportsHistorical: false,
    assetClasses: ["EQUITY", "ETF", "INDEX"],
    exchanges: ["xnys", "xnas", "arcx", "xcbo"],
    homepage: "https://www.cboe.com/delayed_quotes/",
    description: "Cboe's public delayed-quote JSON (last, bid/ask + sizes, open/high/low, previous close, volume, 30-day implied volatility) for a curated US universe of large caps, ETFs and the SPX/NDX/DJI/RUT/VIX indices. 15-minute delayed.",
    rightsNotes: "Delayed data displayed with attribution to Cboe Global Markets; labelled DELAYED everywhere. Not real time.",
    termsUrl: "https://www.cboe.com/about/terms-of-use/",
    sourceFamily: "cboe",
    enabled: true,
  },
  seeds,
  defaultSymbols: SYMBOLS,
  rateLimits: { "cdn.cboe.com": 1 }, // the CDN answers 429 above ~1 req/s
  schedule: { openMs: 120_000, closedMs: 15 * 60_000, weekendMs: 60 * 60_000, exchangeId: "xnys" },
  async poll(ctx) {
    const out: RawObservation[] = [];
    const symbols = ctx.watchedSymbols();
    // Sequential per symbol through the shared limiter; a failed symbol never aborts the batch.
    await Promise.all(
      symbols.map(async (sym) => {
        try {
          const { data, response } = await ctx.http.getJson<Record<string, unknown>>(`${BASE}/${encodeURIComponent(sym)}.json`, { timeoutMs: 15_000 });
          if (response.notModified) return;
          out.push(raw("cboe-delayed-quotes", "cboe", "quote", data, { symbol: sym, status: response.status, ms: response.durationMs }));
        } catch (err) {
          ctx.reportError(err, { symbol: sym });
        }
      }),
    );
    return out;
  },
  normalize(r): NormalizedBatch {
    const p = r.payload as { timestamp?: string; symbol?: string; data?: Record<string, unknown> };
    const d = p?.data;
    if (r.kind !== "quote" || !d || typeof d.symbol !== "string") return { observations: [] };
    const requested = typeof p.symbol === "string" ? p.symbol : d.symbol;
    const index = US_INDICES.find((i) => i.cboe === requested || `^${i.symbol}` === d.symbol);
    const symbol = index ? index.cboe : requested;
    const hint = index ? indexHint(index) : undefined;
    // last_trade_time is exchange local time without offset ("2026-09-11T15:59:59").
    const tradeTs = typeof d.last_trade_time === "string" ? zonedTimeToUtc(d.last_trade_time, ET) : null;
    const snapshotTs = typeof p.timestamp === "string" ? zonedTimeToUtc(p.timestamp.replace(" ", "T"), ET) : null;
    const seq = typeof d.seqno === "number" ? d.seqno : null;
    const observations: NormalizedObservation[] = [];
    const push = (field: NormalizedObservation["field"], v: unknown, ts: number | null, trust: NormalizedObservation["timestampTrust"]) => {
      const n = typeof v === "number" ? v : typeof v === "string" ? Number(v) : NaN;
      if (!Number.isFinite(n)) return;
      if ((field === "BID" || field === "ASK" || field === "LAST_PRICE" || field === "OPEN" || field === "HIGH" || field === "LOW" || field === "PREVIOUS_CLOSE") && n <= 0) return;
      observations.push({ symbol, instrumentHint: hint, field, value: n, currency: "USD", observationType: index ? "INDEX_VALUE" : field === "BID" || field === "ASK" ? "QUOTE" : "TRADE", sourceTimestamp: ts, timestampTrust: trust, sequence: seq, rightsStatus: "DELAYED", realtimeStatus: "DELAYED", meta: { security_type: d.security_type } });
    };
    push("LAST_PRICE", d.current_price, tradeTs, tradeTs ? "EXCHANGE" : "SOURCE");
    push("OPEN", d.open, tradeTs, "EXCHANGE");
    push("HIGH", d.high, tradeTs, "EXCHANGE");
    push("LOW", d.low, tradeTs, "EXCHANGE");
    push("PREVIOUS_CLOSE", d.prev_day_close, tradeTs, "EXCHANGE");
    push("VOLUME", d.volume, tradeTs, "EXCHANGE");
    push("CHANGE", d.price_change, tradeTs, "EXCHANGE");
    push("CHANGE_PERCENT", d.price_change_percent, tradeTs, "EXCHANGE");
    // Quotes (bid/ask) are as of the snapshot, not the last trade.
    push("BID", d.bid, snapshotTs, "SOURCE");
    push("ASK", d.ask, snapshotTs, "SOURCE");
    push("BID_SIZE", d.bid_size, snapshotTs, "SOURCE");
    push("ASK_SIZE", d.ask_size, snapshotTs, "SOURCE");
    push("IMPLIED_VOLATILITY", d.iv30, snapshotTs, "SOURCE");
    return { observations };
  },
  async healthCheck(ctx) {
    const { data } = await ctx.http.getJson<{ data?: { current_price?: number } }>(`${BASE}/SPY.json`);
    return { ok: typeof data?.data?.current_price === "number", detail: `SPY=${data?.data?.current_price}` };
  },
  fixturesDir: "fixtures",
});