1from __future__ import annotations23from datetime import UTC, date, datetime4from pathlib import Path56import pyarrow as pa78from findata_market_firstrate.archive import parse_bar_file9from findata_quality import aggregate_bars, validate_bars10from findata_quality.calendar import xnys11from findata_quality.validators import session_coverage12from findata_schema import Adjustment, AssetClass, Interval, QualityFlag1314FIX = Path(__file__).parent / "fixtures" / "firstrate"15NOW = datetime(2026, 11, 1, 12, 0, tzinfo=UTC)161718def _parse(text: str, symbol="TEST"):19 return parse_bar_file(text.encode(), symbol, AssetClass.stock, Interval.m1, Adjustment.raw, "firstrate", NOW)[0]202122def _flags(t):23 return [set(f) for f in t.column("quality_flags").to_pylist()]242526def test_clean_real_session_has_no_error_flags():27 t = parse_bar_file((FIX / "AAPL_day_1min_UNADJUSTED.txt").read_bytes(), "AAPL", AssetClass.stock, Interval.m1, Adjustment.raw, "firstrate", NOW)[0]28 out, anomalies = validate_bars(t, Interval.m1)29 errors = [a for a in anomalies if a["severity"] == "error"]30 assert errors == []31 assert out.num_rows == 900323334def test_row_level_flags():35 rows = "\n".join([36 "2026-10-30 09:30:00,10,9,8,9.5,100", # high < open → ohlc_inconsistent37 "2026-10-30 09:31:00,10,11,9,10,-5", # negative volume38 "2026-10-30 09:32:30,10,11,9,10,5", # off interval grid39 "2026-10-30 09:33:00,10,11,9,10,5",40 "2026-10-30 09:33:00,10,11,9,10,5", # duplicate41 "2026-10-30 03:59:00,10,11,9,10,5", # before 04:00 ET → outside_session42 "2026-10-30 09:34:00,10,20,9,15,5", # +50 % vs previous close → price_jump43 "2026-10-30 09:35:00,0,0,0,0,5", # nonpositive price44 "2026-11-01 09:30:00,10,11,9,10,5", # Sunday → non_trading_day45 ])46 out, anomalies = validate_bars(_parse(rows), Interval.m1)47 by_ts = {r["timestamp_utc"]: set(r["quality_flags"]) for r in out.to_pylist()}48 f = lambda h, m, d=30: by_ts[datetime(2026, 10, d, h, m, tzinfo=UTC)] # noqa: E73149 assert QualityFlag.ohlc_inconsistent in f(13, 30)50 assert QualityFlag.negative_volume in f(13, 31)51 assert any(QualityFlag.off_interval in v for v in by_ts.values())52 assert QualityFlag.duplicate in f(13, 33)53 assert QualityFlag.outside_session in f(7, 59)54 assert QualityFlag.price_jump in f(13, 34)55 assert QualityFlag.nonpositive_price in f(13, 35)56 assert QualityFlag.non_trading_day in by_ts[datetime(2026, 11, 1, 14, 30, tzinfo=UTC)]57 codes = {a["code"] for a in anomalies}58 assert {"ohlc_inconsistent", "negative_volume", "duplicate", "non_trading_day"} <= codes59 # prices are never altered by validation60 assert out.column("open").to_pylist()[0] == 10.0616263def test_session_calendar_expected_bars_and_early_close():64 cal = xnys()65 assert cal.is_session(date(2026, 10, 30)) and not cal.is_session(date(2026, 11, 1))66 assert not cal.is_session(date(2026, 11, 26)) # Thanksgiving67 assert cal.expected_regular_bars(date(2026, 10, 30)) == 391 # 09:30..16:00 inclusive68 assert cal.expected_regular_bars(date(2026, 11, 27)) == 211 # early close 13:0069 assert cal.expected_regular_bars(date(2026, 11, 1)) == 0707172def test_session_coverage_rows():73 t = parse_bar_file((FIX / "A_day_1min_UNADJUSTED.txt").read_bytes(), "A", AssetClass.stock, Interval.m1, Adjustment.raw, "firstrate", NOW)[0]74 cov = session_coverage(t, Interval.m1)75 assert len(cov) == 1 and cov[0]["symbol"] == "A" and cov[0]["bar_count"] == 390 and cov[0]["expected_bars"] == 391767778def test_aggregate_5m_matches_manual_expectation():79 t = parse_bar_file((FIX / "AAPL_day_1min_UNADJUSTED.txt").read_bytes(), "AAPL", AssetClass.stock, Interval.m1, Adjustment.raw, "firstrate", NOW)[0]80 agg = aggregate_bars(t, Interval.m5)81 import polars as pl82 one = pl.from_arrow(t).sort("timestamp_utc")83 five = pl.from_arrow(agg).sort("timestamp_utc")84 # 09:30 ET bucket = 13:30 UTC covers 13:30..13:3485 win = one.filter((pl.col("timestamp_utc") >= datetime(2026, 10, 30, 13, 30, tzinfo=UTC)) & (pl.col("timestamp_utc") < datetime(2026, 10, 30, 13, 35, tzinfo=UTC)))86 b = five.filter(pl.col("timestamp_utc") == datetime(2026, 10, 30, 13, 30, tzinfo=UTC)).row(0, named=True)87 assert b["open"] == win["open"][0] and b["close"] == win["close"][-1]88 assert b["high"] == win["high"].max() and b["low"] == win["low"].min()89 assert abs(b["volume"] - win["volume"].sum()) < 1e-6 and b["trade_count"] == 590 assert b["interval"] == "5m" and b["session_date"] == date(2026, 10, 30)91 assert abs(five["volume"].sum() - one["volume"].sum()) < 1e-6 # volume conserved92 assert five.height <= 16 * 12 # 04:00–20:00 → ≤ 192 buckets, none manufactured93 assert all(ts.minute % 5 == 0 for ts in five["timestamp_utc"].to_list())949596def test_aggregate_buckets_follow_wall_clock_across_dst():97 # 09:30 EST (14:30 UTC) and 09:30 EDT (13:30 UTC) must each start a bucket98 t = _parse("2026-03-06 09:31:00,1,1,1,1,1\n2026-03-06 09:34:00,2,2,2,2,1\n2026-03-09 09:31:00,3,3,3,3,1\n2026-03-09 09:34:00,4,4,4,4,1\n")99 agg = aggregate_bars(t, Interval.m5)100 ts = agg.column("timestamp_utc").to_pylist()101 assert ts == [datetime(2026, 3, 6, 14, 30, tzinfo=UTC), datetime(2026, 3, 9, 13, 30, tzinfo=UTC)]102 assert agg.column("open").to_pylist() == [1.0, 3.0] and agg.column("close").to_pylist() == [2.0, 4.0]103104105def test_aggregate_zero_volume_bucket_has_null_vwap():106 t = _parse("2026-10-30 09:30:00,1,1,1,1,0\n")107 agg = aggregate_bars(t, Interval.m5)108 assert agg.column("vwap")[0].as_py() is None109