"""Repository for cached OHLCV bars and their coverage ranges."""
import re
from collections.abc import Sequence
from datetime import UTC, datetime, timedelta
from sqlalchemy import delete, select
from sqlalchemy.orm import Session, sessionmaker
from trader.domain import Bar
from trader.persistence.models import BarCoverage, CachedBar
__all__ = ["BarRepository", "bar_width"]
#: Seconds per yfinance interval suffix. `mo` is deliberately absent: a month
#: is not a fixed duration, so no single `timedelta` is correct for it.
_UNIT_SECONDS = {"m": 60, "h": 3600, "d": 86400, "wk": 604800}
_INTERVAL_PATTERN = re.compile(r"(\d+)(m|h|d|wk)")
#: The grid coverage is snapped to. The origin's *weekday* matters: flooring by
#: a week lands on whatever weekday the origin is, and yfinance dates weekly
#: bars on Mondays. 1970-01-01 is a Thursday, so the epoch would silently align
#: weekly coverage three days off the bars it describes — the same
#: grid-misalignment bug this module exists to prevent. 1969-12-29 is the
#: Monday before the epoch, and is still midnight UTC so daily, hourly and
#: minute flooring are unaffected.
_GRID_ORIGIN = datetime(1969, 12, 29, tzinfo=UTC)
[docs]
def bar_width(interval: str) -> timedelta:
"""How long one bar of `interval` spans.
Used for two things that must agree: snapping coverage onto the bar grid,
and deciding whether two coverage ranges abut. They were allowed to
disagree until 2026-08-04, when coverage recorded at instant precision
against midnight-dated daily bars made `missing_ranges` answer differently
depending on the time of day, and a live limit order was priced off the
previous day's close.
Raises:
ValueError: the interval is not one this repository can reason about.
A guessed width silently corrupts coverage, which is worse than
refusing, so there is no default.
"""
match = _INTERVAL_PATTERN.fullmatch(interval)
if match is None:
raise ValueError(
f"Unsupported bar interval {interval!r}: expected a count followed "
"by m, h, d, or wk — for example '15m', '1h', '1d'."
)
count, unit = match.groups()
return timedelta(seconds=int(count) * _UNIT_SECONDS[unit])
def _snap(moment: datetime, interval: str) -> datetime:
"""Floor `moment` onto the bar grid for `interval`."""
width = bar_width(interval)
return moment - ((moment - _GRID_ORIGIN) % width)
[docs]
class BarRepository:
"""Reads and writes cached bars, and tracks which ranges were fetched."""
def __init__(self, session_factory: sessionmaker[Session]) -> None:
self._session_factory = session_factory
[docs]
def save_bars(
self,
symbol: str,
interval: str,
bars: Sequence[Bar],
source: str | None = None,
) -> int:
"""Insert bars, skipping any timestamp already stored. Returns inserted count.
`source` (issue #41) tags every row this call inserts — `None` for the
live path (`BarCache`, which never passes it), a vendor tag such as
`seed_import.SOURCE_YFINANCE_SEED` for `trader seed-bars`. It is
per-call, not per-bar: one call always describes bars pulled from one
file of one known provenance, so there is nothing to gain from letting
individual `Bar`s disagree, and the domain `Bar` type itself carries
no such field — `source` is a fact about how the row entered this
table, not a fact yfinance or Interactive Brokers ever reports back.
"""
if not bars:
return 0
with self._session_factory() as session:
seen = set(
session.scalars(
select(CachedBar.timestamp).where(
CachedBar.symbol == symbol, CachedBar.interval == interval
)
).all()
)
new: list[Bar] = []
for b in bars:
# Dedupe against the input too: two copies of one timestamp in a
# single call would both pass a DB-only check and then collide on
# the unique constraint at commit.
if b.timestamp not in seen:
seen.add(b.timestamp)
new.append(b)
session.add_all(
CachedBar(
symbol=symbol,
interval=interval,
timestamp=b.timestamp,
open=b.open,
high=b.high,
low=b.low,
close=b.close,
volume=b.volume,
source=source,
)
for b in new
)
session.commit()
return len(new)
[docs]
def delete_bars(
self, symbol: str, interval: str, start: datetime, end: datetime
) -> int:
"""Delete cached bars in the window. Returns how many were removed.
Required for refresh: `save_bars` deliberately skips timestamps it
already holds, so re-fetching revised data would otherwise be a no-op
that still stamps a fresh `fetched_at`.
Scoped to `source IS NULL` (issue #41) — a seeded row (non-`NULL`
`source`) is never deleted by this method, at any call site,
including the trading path's automatic tail refresh
(`BarCache._refresh_tail_or_degrade`) and the operator-invoked
`trader fetch-bars --refresh`. Seed data is imported once, from a
file that will not change; the trading path's own tail-refresh
policy exists specifically because *its* bars can still be corrected
by a later fetch, which is not true of a historical import. Without
this guard, a live refresh window that happened to overlap an
imported symbol's dates would silently replace validated backtest
input with a differently-adjusted live fetch — exactly what the seed
importer exists to avoid needing in the first place.
"""
with self._session_factory() as session:
result = session.execute(
delete(CachedBar).where(
CachedBar.symbol == symbol,
CachedBar.interval == interval,
CachedBar.timestamp >= start,
CachedBar.timestamp <= end,
CachedBar.source.is_(None),
)
)
session.commit()
return result.rowcount
[docs]
def load_bars(
self, symbol: str, interval: str, start: datetime, end: datetime
) -> list[Bar]:
"""Return cached bars in the window, oldest first."""
statement = (
select(CachedBar)
.where(
CachedBar.symbol == symbol,
CachedBar.interval == interval,
CachedBar.timestamp >= start,
CachedBar.timestamp <= end,
)
.order_by(CachedBar.timestamp.asc())
)
with self._session_factory() as session:
rows = session.scalars(statement).all()
return [
Bar(
symbol=r.symbol,
timestamp=r.timestamp,
open=r.open,
high=r.high,
low=r.low,
close=r.close,
volume=r.volume,
)
for r in rows
]
[docs]
def covered_ranges(
self, symbol: str, interval: str
) -> list[tuple[datetime, datetime]]:
"""Merged, sorted ranges known to have been fetched."""
statement = (
select(BarCoverage)
.where(BarCoverage.symbol == symbol, BarCoverage.interval == interval)
.order_by(BarCoverage.start_date.asc())
)
with self._session_factory() as session:
rows = session.scalars(statement).all()
return _merge([(r.start_date, r.end_date) for r in rows], bar_width(interval))
[docs]
def record_coverage(
self, symbol: str, interval: str, start: datetime, end: datetime
) -> None:
"""Record a fetched range, merging it into any it overlaps or abuts.
Read, delete, and insert happen in one transaction: splitting them lets
two concurrent callers each rewrite the row set from a stale read, and
the later commit silently discards the other's range.
"""
now = datetime.now(UTC)
start = _snap(start, interval)
end = _snap(end, interval)
with self._session_factory() as session:
rows = session.scalars(
select(BarCoverage)
.where(BarCoverage.symbol == symbol, BarCoverage.interval == interval)
.order_by(BarCoverage.start_date.asc())
).all()
merged = _merge(
[(r.start_date, r.end_date) for r in rows] + [(start, end)],
bar_width(interval),
)
session.execute(
delete(BarCoverage).where(
BarCoverage.symbol == symbol, BarCoverage.interval == interval
)
)
session.add_all(
BarCoverage(
symbol=symbol,
interval=interval,
start_date=s,
end_date=e,
fetched_at=now,
)
for s, e in merged
)
session.commit()
[docs]
def missing_ranges(
self, symbol: str, interval: str, start: datetime, end: datetime
) -> list[tuple[datetime, datetime]]:
"""Sub-ranges of [start, end] that have never been fetched.
The window is snapped to the bar grid first, so the answer depends only
on which bars are wanted and not on the clock time of the call.
Adjacency is one bar-width, not one day. Two intraday ranges an hour
apart are genuinely not contiguous, and merging them would report the
hole between as covered — after which nothing would ever re-fetch it.
"""
adjacency = bar_width(interval)
start = _snap(start, interval)
end = _snap(end, interval)
gaps: list[tuple[datetime, datetime]] = []
cursor = start
for covered_start, covered_end in self.covered_ranges(symbol, interval):
if covered_end < cursor:
continue
if covered_start > end:
break
if covered_start > cursor:
gaps.append((cursor, min(covered_start - adjacency, end)))
cursor = max(cursor, covered_end + adjacency)
if cursor > end:
return gaps
if cursor <= end:
gaps.append((cursor, end))
return gaps
def _merge(
ranges: list[tuple[datetime, datetime]], adjacency: timedelta
) -> list[tuple[datetime, datetime]]:
"""Merge overlapping or adjacent ranges into a minimal sorted list."""
if not ranges:
return []
ordered = sorted(ranges)
merged = [ordered[0]]
for start, end in ordered[1:]:
last_start, last_end = merged[-1]
if start <= last_end + adjacency:
merged[-1] = (last_start, max(last_end, end))
else:
merged.append((start, end))
return merged