Chris Amow
fced18528b
History was only fetched by the startup seed, so a Schwab outage stayed a hole until the next deploy — Sept 9 to 24 after a refresh token expired. A reconnect more than two minutes past the last bar now fetches the gap from Yahoo, fills empty buckets only, refolds the live forming buckets, rebuilds levels without alerting, and resyncs every socket. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
69 lines
2.6 KiB
Python
69 lines
2.6 KiB
Python
from app.bars.models import Bar, Timeframe
|
|
from app.bars.store import InMemoryBarStore
|
|
|
|
|
|
def bar(t, close=1):
|
|
return Bar(Timeframe.M1, t, 1, 2, 0, close, 10, False, "ES=F", "replay")
|
|
|
|
|
|
def test_store_replaces_forming_bar_and_bounds_history():
|
|
store = InMemoryBarStore(2)
|
|
store.put(bar(60))
|
|
store.put(bar(60, 2))
|
|
store.put(bar(120))
|
|
store.put(bar(180))
|
|
|
|
assert [value.t for value in store.get(Timeframe.M1)] == [120, 180]
|
|
assert store.get(Timeframe.M1, 1)[0].t == 180
|
|
|
|
|
|
def test_a_closed_bar_replaces_its_bucket_behind_the_tail():
|
|
# Ticks open the next minute before the exchange's own bar for the previous
|
|
# one arrives. Matching only the tail dropped it, leaving the tick-built
|
|
# approximation — with its partial volume — in place permanently.
|
|
from app.bars.models import Bar, Timeframe
|
|
from app.bars.store import InMemoryBarStore
|
|
|
|
store = InMemoryBarStore(100)
|
|
common = dict(tf=Timeframe.M1, symbol="/ES", source="schwab")
|
|
store.put(Bar(t=60, o=1, h=2, l=1, c=2, v=5, closed=False, **common))
|
|
store.put(Bar(t=120, o=2, h=3, l=2, c=3, v=1, closed=False, **common))
|
|
# The authoritative bar for the first minute, arriving late.
|
|
store.put(Bar(t=60, o=1, h=9, l=1, c=4, v=400, closed=True, **common))
|
|
|
|
held = store.get(Timeframe.M1)
|
|
assert [b.t for b in held] == [60, 120]
|
|
assert held[0].closed is True
|
|
assert held[0].v == 400, "the exchange's volume must win over the tick estimate"
|
|
|
|
|
|
def test_a_tick_cannot_overwrite_a_settled_bar():
|
|
from app.bars.models import Bar, Timeframe
|
|
from app.bars.store import InMemoryBarStore
|
|
|
|
store = InMemoryBarStore(100)
|
|
common = dict(tf=Timeframe.M1, symbol="/ES", source="schwab")
|
|
store.put(Bar(t=60, o=1, h=9, l=1, c=4, v=400, closed=True, **common))
|
|
store.put(Bar(t=60, o=1, h=2, l=1, c=2, v=5, closed=False, **common))
|
|
|
|
held = store.get(Timeframe.M1)[0]
|
|
assert held.closed is True and held.v == 400
|
|
|
|
|
|
def test_a_backfilled_hole_lands_behind_live_bars_without_replacing_them():
|
|
# After an outage the live stream is already newer than the hole, so put()
|
|
# dropped every recovered bar and fifteen missed days stayed missing.
|
|
store = InMemoryBarStore(4)
|
|
store.put(bar(60))
|
|
store.put(bar(600, close=7))
|
|
|
|
added = store.fill([bar(120), bar(180), bar(600, close=99)])
|
|
|
|
assert added == 2
|
|
assert [value.t for value in store.get(Timeframe.M1)] == [60, 120, 180, 600]
|
|
assert store.get(Timeframe.M1)[-1].c == 7, "the live source's bar must win"
|
|
|
|
store.fill([bar(240)])
|
|
assert [value.t for value in store.get(Timeframe.M1)] == [120, 180, 240, 600], (
|
|
"the cap keeps the newest history"
|
|
)
|