Implement M0 Yahoo market data and replay

app/__init__.py

@@ -0,0 +1 @@
+"""Application package."""

app/bars/__init__.py

@@ -0,0 +1 @@
+"""Bar models, storage, and aggregation."""

app/bars/models.py

@@ -0,0 +1,63 @@
+from dataclasses import asdict, dataclass
+from enum import Enum
+from typing import Any
+
+
+class Timeframe(str, Enum):
+    M1 = "1m"
+    M2 = "2m"
+    M5 = "5m"
+    M15 = "15m"
+    M30 = "30m"
+    H1 = "1h"
+    H4 = "4h"
+    D1 = "1d"
+
+    @property
+    def seconds(self) -> int:
+        seconds = {
+            self.M1: 60,
+            self.M2: 120,
+            self.M5: 300,
+            self.M15: 900,
+            self.M30: 1800,
+            self.H1: 3600,
+            self.H4: 14400,
+        }
+        if self is self.D1:
+            raise ValueError("1d is session-defined, not a fixed number of seconds")
+        return seconds[self]
+
+
+@dataclass(slots=True)
+class Bar:
+    tf: Timeframe
+    t: int
+    o: float
+    h: float
+    l: float
+    c: float
+    v: int
+    closed: bool
+    symbol: str
+    source: str
+
+    def to_dict(self) -> dict[str, Any]:
+        value = asdict(self)
+        value["tf"] = self.tf.value
+        return value
+
+    @classmethod
+    def from_dict(cls, value: dict[str, Any]) -> "Bar":
+        return cls(
+            tf=Timeframe(value["tf"]),
+            t=int(value["t"]),
+            o=float(value["o"]),
+            h=float(value["h"]),
+            l=float(value["l"]),
+            c=float(value["c"]),
+            v=int(value.get("v") or 0),
+            closed=bool(value["closed"]),
+            symbol=str(value["symbol"]),
+            source=str(value["source"]),
+        )

app/market/__init__.py

@@ -0,0 +1 @@
+"""Pluggable market data sources."""

app/market/base.py

@@ -0,0 +1,22 @@
+from collections.abc import AsyncIterator
+from typing import Protocol
+
+from app.bars.models import Bar, Timeframe
+
+
+class MarketDataSource(Protocol):
+    name: str
+
+    def supports_history(self) -> bool: ...
+
+    async def history(
+        self,
+        symbol: str,
+        tf: Timeframe,
+        start: int | None,
+        end: int | None,
+    ) -> list[Bar]: ...
+
+    def supports_stream(self) -> bool: ...
+
+    def stream(self, symbol: str) -> AsyncIterator[Bar]: ...

app/market/recorder.py

@@ -0,0 +1,20 @@
+import json
+import time
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+from app.bars.models import Bar
+
+
+class Recorder:
+    def __init__(self, path: str | Path):
+        self.path = Path(path)
+
+    async def record(self, stream: AsyncIterator[Bar]) -> AsyncIterator[Bar]:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with self.path.open("a", encoding="utf-8") as tape:
+            async for bar in stream:
+                row = {"arrival_t": time.time(), "bar": bar.to_dict()}
+                tape.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n")
+                tape.flush()
+                yield bar

app/market/replay.py

@@ -0,0 +1,54 @@
+import asyncio
+import json
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+from app.bars.models import Bar, Timeframe
+
+
+class ReplaySource:
+    name = "replay"
+
+    def __init__(self, path: str | Path, speed: float = 0):
+        self.path = Path(path)
+        self.speed = speed
+
+    def supports_history(self) -> bool:
+        return True
+
+    def supports_stream(self) -> bool:
+        return True
+
+    def _rows(self) -> list[dict]:
+        if not self.path.exists():
+            return []
+        with self.path.open(encoding="utf-8") as tape:
+            return [json.loads(line) for line in tape if line.strip()]
+
+    async def history(
+        self,
+        symbol: str,
+        tf: Timeframe,
+        start: int | None = None,
+        end: int | None = None,
+    ) -> list[Bar]:
+        return [
+            bar
+            for row in self._rows()
+            if (bar := Bar.from_dict(row["bar"])).tf is tf
+            and (not symbol or bar.symbol == symbol)
+            and (start is None or bar.t >= start)
+            and (end is None or bar.t < end)
+        ]
+
+    async def stream(self, symbol: str) -> AsyncIterator[Bar]:
+        previous_arrival: float | None = None
+        for row in self._rows():
+            bar = Bar.from_dict(row["bar"])
+            if symbol and bar.symbol != symbol:
+                continue
+            arrival = float(row["arrival_t"])
+            if self.speed > 0 and previous_arrival is not None:
+                await asyncio.sleep(max(0, arrival - previous_arrival) / self.speed)
+            previous_arrival = arrival
+            yield bar

app/market/yahoo.py

@@ -0,0 +1,116 @@
+import asyncio
+from collections.abc import AsyncIterator
+from typing import Any
+
+import httpx
+
+from app.bars.models import Bar, Timeframe
+
+YAHOO_CHART_URL = "https://query1.finance.yahoo.com/v8/finance/chart/{symbol}"
+MAX_1M_WINDOW_SECONDS = 8 * 24 * 60 * 60
+
+
+def parse_chart(payload: dict[str, Any], tf: Timeframe, source: str = "yahoo") -> list[Bar]:
+    chart = payload.get("chart", {})
+    if chart.get("error"):
+        raise ValueError(f"Yahoo chart error: {chart['error']}")
+    results = chart.get("result") or []
+    if not results:
+        return []
+
+    result = results[0]
+    timestamps = result.get("timestamp") or []
+    quotes = ((result.get("indicators") or {}).get("quote") or [{}])[0]
+    symbol = str((result.get("meta") or {}).get("symbol") or "")
+    arrays = [quotes.get(key) or [] for key in ("open", "high", "low", "close", "volume")]
+    bars: list[Bar] = []
+    for values in zip(timestamps, *arrays, strict=False):
+        t, open_, high, low, close, volume = values
+        if any(value is None for value in (t, open_, high, low, close)):
+            continue
+        bars.append(
+            Bar(
+                tf=tf,
+                t=int(t),
+                o=float(open_),
+                h=float(high),
+                l=float(low),
+                c=float(close),
+                v=int(volume or 0),
+                closed=True,
+                symbol=symbol,
+                source=source,
+            )
+        )
+    return bars
+
+
+class YahooSource:
+    name = "yahoo"
+
+    def __init__(self, poll_seconds: float = 20, client: httpx.AsyncClient | None = None):
+        self.poll_seconds = poll_seconds
+        self._client = client
+
+    def supports_history(self) -> bool:
+        return True
+
+    def supports_stream(self) -> bool:
+        return True
+
+    async def _fetch(self, symbol: str, params: dict[str, str | int]) -> dict[str, Any]:
+        owns_client = self._client is None
+        client = self._client or httpx.AsyncClient(
+            headers={"User-Agent": "Mozilla/5.0 chart.amow.com"}, timeout=30
+        )
+        try:
+            response = await client.get(YAHOO_CHART_URL.format(symbol=symbol), params=params)
+            response.raise_for_status()
+            return response.json()
+        finally:
+            if owns_client:
+                await client.aclose()
+
+    async def history(
+        self,
+        symbol: str,
+        tf: Timeframe,
+        start: int | None = None,
+        end: int | None = None,
+        *,
+        range_: str | None = None,
+    ) -> list[Bar]:
+        if tf not in (Timeframe.M1, Timeframe.H1):
+            raise ValueError("YahooSource history supports only 1m and 1h inputs")
+        interval = tf.value
+        if range_ is not None:
+            return parse_chart(
+                await self._fetch(symbol, {"interval": interval, "range": range_}), tf
+            )
+        if start is None or end is None:
+            raise ValueError("start/end or range_ is required")
+        if end <= start:
+            return []
+
+        window = MAX_1M_WINDOW_SECONDS if tf is Timeframe.M1 else end - start
+        by_time: dict[int, Bar] = {}
+        cursor = start
+        while cursor < end:
+            window_end = min(cursor + window, end)
+            payload = await self._fetch(
+                symbol,
+                {"interval": interval, "period1": cursor, "period2": window_end},
+            )
+            by_time.update((bar.t, bar) for bar in parse_chart(payload, tf))
+            cursor = window_end
+        return [by_time[t] for t in sorted(by_time)]
+
+    async def stream(self, symbol: str) -> AsyncIterator[Bar]:
+        last_emitted = -1
+        while True:
+            bars = await self.history(symbol, Timeframe.M1, range_="1d")
+            for bar in bars:
+                if bar.t > last_emitted:
+                    yield bar
+                    last_emitted = bar.t
+            await asyncio.sleep(self.poll_seconds)

requirements-dev.txt

@@ -0,0 +1,2 @@
+pytest
+pytest-asyncio

requirements.txt

@@ -1,2 +1,4 @@
 fastapi
 uvicorn[standard]
+httpx
+pydantic-settings

tests/test_replay.py

@@ -0,0 +1,31 @@
+import json
+
+import pytest
+
+from app.bars.models import Bar, Timeframe
+from app.market.recorder import Recorder
+from app.market.replay import ReplaySource
+
+
+def sample_bars():
+    return [
+        Bar(Timeframe.M1, 100 + i * 60, 1, 2, 0.5, 1.5, 10, True, "ES=F", "yahoo")
+        for i in range(2)
+    ]
+
+
+@pytest.mark.asyncio
+async def test_recorded_tape_replays_identically(tmp_path):
+    expected = sample_bars()
+
+    async def source():
+        for bar in expected:
+            yield bar
+
+    tape = tmp_path / "tape.jsonl"
+    recorded = [bar async for bar in Recorder(tape).record(source())]
+    replayed = [bar async for bar in ReplaySource(tape).stream("ES=F")]
+
+    assert recorded == expected
+    assert replayed == expected
+    assert len([json.loads(line) for line in tape.read_text().splitlines()]) == 2

tests/test_yahoo.py

@@ -0,0 +1,40 @@
+import json
+from pathlib import Path
+
+import httpx
+import pytest
+
+from app.bars.models import Timeframe
+from app.market.yahoo import MAX_1M_WINDOW_SECONDS, YahooSource, parse_chart
+
+
+FIXTURE = Path(__file__).parent / "fixtures" / "yahoo_es_1h.json"
+
+
+def test_parser_filters_null_ohlc_rows():
+    payload = json.loads(FIXTURE.read_text())
+    raw_count = len(payload["chart"]["result"][0]["timestamp"])
+    bars = parse_chart(payload, Timeframe.H1)
+
+    assert len(bars) == raw_count - 1
+    assert all(None not in (bar.o, bar.h, bar.l, bar.c) for bar in bars)
+    assert all(bar.source == "yahoo" and bar.symbol == "ES=F" for bar in bars)
+
+
+@pytest.mark.asyncio
+async def test_one_minute_history_is_windowed_and_deduplicated():
+    payload = json.loads(FIXTURE.read_text())
+    calls = []
+
+    async def handler(request: httpx.Request):
+        calls.append(request)
+        return httpx.Response(200, json=payload)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        source = YahooSource(client=client)
+        bars = await source.history(
+            "ES=F", Timeframe.M1, 0, MAX_1M_WINDOW_SECONDS * 2 + 1
+        )
+
+    assert len(calls) == 3
+    assert len(bars) == len({bar.t for bar in parse_chart(payload, Timeframe.M1)})