1"""Full vendor catalog: new layouts (index without volume, futures contracts, option chains), no-data responses,2hot/cold lake roots, the complete plan, and the options task through the vendor mock."""3from __future__ import annotations45import io6import zipfile7from datetime import UTC, date, datetime89import httpx10import pytest11from findata_market_firstrate import (12 FRDClient,13 FRDNoData,14 build_full_plan,15 dataset_id_for,16 parse_options_file,17 ticker_from_filename,18)19from findata_market_firstrate.archive import parse_bar_file20from findata_quality import validate_bars21from findata_schema import OPTIONS_SCHEMA, Adjustment, AssetClass, Interval2223from tests.conftest import TEST_USERID2425NOW = datetime(2026, 11, 1, 12, 0, tzinfo=UTC)262728def test_index_without_volume_and_fx_two_fields():29 t, raw = parse_bar_file(b"2026-10-29 04:00:00,1136.43,1137.03,1135.97,1135.97\n2026-10-29 04:01:00,1135.89,1135.89,1134.56,1134.83\n",30 "AEX", AssetClass.index, Interval.m1, Adjustment.raw, "firstrate", NOW)31 assert raw == 2 and t.num_rows == 2 and t.column("volume").to_pylist() == [0.0, 0.0]32 out, anomalies = validate_bars(t, Interval.m1, asset_class="index")33 assert all("zero_volume" not in f for f in out.column("quality_flags").to_pylist())34 t, _ = parse_bar_file(b"20261026,00:02:00,3.49536933,3.49536933,3.49536933,3.49536933,1\n", "AUDBRL", AssetClass.fx, Interval.m1, Adjustment.raw, "firstrate", NOW)35 assert t.column("timestamp_utc")[0].as_py() == datetime(2026, 10, 26, 4, 2, tzinfo=UTC) # 00:02 EDT36 out, anomalies = validate_bars(t, Interval.m1, asset_class="fx")37 assert not any(a["code"] in ("outside_session", "non_trading_day") for a in anomalies) # 24h market: no session checks38 t, _ = parse_bar_file(b"2026-10-30 00:00:00,119.071,119.46,119.04,119.045,208\n", "AAVE", AssetClass.crypto, Interval.m1, Adjustment.raw, "firstrate", NOW, source_tz="UTC")39 assert t.column("timestamp_utc")[0].as_py() == datetime(2026, 10, 30, 0, 0, tzinfo=UTC)404142def test_contract_filenames_and_daily_open_interest():43 assert ticker_from_filename("A6_F25_1day.txt", contract=True) == "A6_F25"44 assert ticker_from_filename("ES_H25_1min.txt", contract=True) == "ES_H25"45 assert ticker_from_filename("ES_full_1min_contin_adj_ratio.txt") == "ES"46 t, _ = parse_bar_file(b"2024-09-11,0.6669,0.6679,0.6669,0.6679,10,5\n", "A6_F25", AssetClass.futures, Interval.d1, Adjustment.raw, "firstrate", NOW)47 assert t.num_rows == 1 and t.column("session_date")[0].as_py() == date(2024, 9, 11)484950def test_option_chain_parser():51 data = b"2026-10-01,33,2026-10-16,c,8.14,7.3,8.8,0.0,1.102138,0,0,1.0,0.0,1e-05,0.0,0.0\n2026-10-01,34,2026-10-16,p,7.14,6.4,7.9,0.0,1.048606,12,3,-0.4,0.01,1e-05,-0.02,0.0\n"52 t, raw = parse_options_file(data, "AAL", "firstrate", NOW)53 assert raw == 2 and t.num_rows == 2 and t.schema.equals(OPTIONS_SCHEMA)54 r = t.to_pylist()[1]55 assert r["underlying"] == "AAL" and r["call_put"] == "p" and r["strike"] == 34.0 and r["expiry"] == date(2026, 10, 16) and r["open_interest"] == 12.0565758def test_no_data_response_is_distinguished_from_errors(tmp_path):59 def handler(req: httpx.Request) -> httpx.Response:60 return httpx.Response(200, text="No update available", headers={"content-type": "text/plain", "content-length": "19"})61 c = FRDClient(TEST_USERID, min_interval_s=0, transport=httpx.MockTransport(handler))62 with pytest.raises(FRDNoData):63 c.download("full_update", {"type": "stock", "period": "week", "timeframe": "1day", "adjustment": "adj_split"}, tmp_path / "x.zip")646566def test_dataset_ids_and_full_plan():67 assert dataset_id_for("stock", "1min", "adj_splitdiv") == "market_bars.stock.1m.split_div"68 assert dataset_id_for("etf", "1day", "UNADJUSTED") == "market_bars.etf.1d.raw"69 assert dataset_id_for("futures", "30min", "contin_adj_absolute") == "market_bars.futures.30m.continuous_absolute"70 assert dataset_id_for("futures", "1min", None, contracts=True) == "market_bars.futures_contracts.1m.raw"71 assert dataset_id_for("options", "1day", None) == "options.eod"72 plan = build_full_plan(today=date(2026, 11, 1))73 assert len(plan) == 79874 assert plan[0]["dataset_id"] == "market_bars.stock.1m.raw" and plan[0]["params"]["ticker_range"] == "A"75 assert sum(1 for j in plan if j["kind"] == "options") == 67 + 176 assert len({j["dedupe_key"] for j in plan}) == len(plan)77 assert all(j["priority"] >= 10 for j in plan) and plan[-1]["kind"] == "options"78 small = build_full_plan(types=["index"], include_options=False, include_delisted=False, today=date(2026, 11, 1))79 assert len(small) == 5808182def test_cold_root_placement(tmp_path, monkeypatch):83 from findata_ingest.lake import Lake84 cold = tmp_path / "ssd"85 cold.mkdir()86 with pytest.raises(RuntimeError):87 Lake(tmp_path / "hot", cold) # marker missing → refuse (volume may be unmounted)88 (cold / ".findata-cold-root").write_text("x")89 lake = Lake(tmp_path / "hot", cold)90 assert lake.root_for("market_bars.stock.1m.raw") == tmp_path / "hot"91 assert lake.root_for("options.eod") == cold and lake.root_for("market_bars.stock.1m.split_div") == cold92 assert lake.raw_dir("firstrate", "x").is_relative_to(cold)93 assert lake.abs("curated/options.eod/base/underlying=A/period=2024q1/p.parquet").is_relative_to(cold)94 assert lake.abs("curated/market_bars.stock.1m.raw/delta/session_date=2026-10-30/p.parquet").is_relative_to(tmp_path / "hot")95 assert lake.rel(cold / "raw" / "a.zip") == "raw/a.zip"969798def test_options_task_with_vendor_mock(vendor, lake, log, monkeypatch):99 from findata_core.db import connection, fetch_one100 from findata_ingest import queue101 from findata_ingest.tasks import options as O102103 def opt_zip() -> bytes:104 buf = io.BytesIO()105 with zipfile.ZipFile(buf, "w") as zf:106 zf.writestr(zipfile.ZipInfo("AAL_month_option_chain.txt", (2026, 10, 30, 0, 0, 0)), "2026-10-01,33,2026-10-16,c,8.14,7.3,8.8,0.0,1.1,0,0,1.0,0.0,1e-05,0.0,0.0\n2026-10-02,33,2026-10-16,c,8.0,7.2,8.7,0.0,1.1,1,2,1.0,0.0,1e-05,0.0,0.0\n")107 zf.writestr(zipfile.ZipInfo("SPY_month_option_chain.txt", (2026, 10, 30, 0, 0, 0)), "2026-10-01,500,2026-10-17,p,1.5,1.4,1.6,0.2,0.21,100,50,-0.3,0.01,0.1,-0.05,0.0\n")108 return buf.getvalue()109 vendor.archive_override = opt_zip()110 jid = queue.enqueue("options", "options.eod", {"period": "month"})111 job = queue.claim_next(["options"])112 r = O.run(job, lake, log)113 queue.finish(job, True, r)114 assert r["underlyings"] == 2 and r["rows"] == 3 and r["files"] == 2115 with connection() as conn:116 rows = conn.execute("SELECT symbol_min, date_min, date_max, row_count FROM lake_files WHERE dataset_id='options.eod' AND version_removed IS NULL ORDER BY 1").fetchall()117 assert [(x["symbol_min"], x["row_count"]) for x in rows] == [("AAL", 2), ("SPY", 1)]118 assert fetch_one("SELECT status FROM datasets WHERE id='options.eod'")["status"] == "active"119 # the month archive re-run supersedes the `latest` partition (no duplicates)120 job = queue.claim_next(["options"]) or (queue.enqueue("options", "options.eod", {"period": "month"}) and queue.claim_next(["options"]))121 r2 = O.run(job, lake, log)122 queue.finish(job, True, r2)123 with connection() as conn:124 live = conn.execute("SELECT count(*) AS n, sum(row_count) AS rows FROM lake_files WHERE dataset_id='options.eod' AND version_removed IS NULL").fetchone()125 assert live["n"] == 2 and live["rows"] == 3126 vendor.archive_override = None127