momentum devandClaude Fable 5
f6999c7f6a
premarket_eval's shadow enrichment call site queried session_date = current_date (wrong on UTC-server Postgres at night, so it never matched any symbols) and shared one try-block for float_cache + dilution filings (a float_cache failure silently skipped the dilution refresh too). Fix: compute the ET session date in Python, split into two independent try-blocks, lift into a testable _refresh_shadow_enrichment(con, session_date). float_cache now applies its own DDL unconditionally on use instead of only when an API key is present. Adds iv_dilution_gate (additive, mirrors iv_regime_gate's keep-or-zero shape keyed on symbol instead of session_date, fail-open on unfetched symbols) plus its once-per-run preload, and registers the pre-registered dilution_veto_v1 experiment via the existing experiments/*.json self-registering cohort. Live veto stays OFF; this only wires the shadow measurement. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
52 lines
2.1 KiB
Python
52 lines
2.1 KiB
Python
"""research/ingestion/float_cache.py (plan 016 step 2) — the module's own
|
|
``CREATE TABLE IF NOT EXISTS float_cache`` DDL must apply on use so a fresh
|
|
schema (like a first-boot production DB) never throws UndefinedTable, even
|
|
when there's no FMP_API_KEY or no symbols to fetch. Runs against an ISOLATED
|
|
throwaway Postgres schema (EventStore ":memory:"), never production, and
|
|
makes no HTTP call (empty symbol list).
|
|
|
|
Both FMP env var names are explicitly cleared via monkeypatch before the
|
|
no-key test: an unrelated test elsewhere (test_admin_effective_config.py,
|
|
via ``import admin_api``) triggers admin_api's own module-level bare
|
|
``load_dotenv()`` during pytest COLLECTION, which walks up from cwd and can
|
|
pick up the outer repo's real .env (this worktree is nested under it) —
|
|
polluting os.environ for the rest of the session. Pre-existing, out of scope
|
|
for this plan; this test must not depend on ambient environment either way."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from storage.event_store import EventStore
|
|
from research.ingestion.float_cache import get_float, refresh_float_cache
|
|
|
|
|
|
def test_refresh_creates_table_and_is_a_noop_on_empty_symbols():
|
|
s = EventStore(":memory:")
|
|
try:
|
|
out = refresh_float_cache(s.con, [], api_key="x")
|
|
assert out == {"fetched": 0, "skipped_fresh": 0, "failed": 0, "no_key": False}
|
|
rows = s.con.execute("SELECT count(*) FROM float_cache").fetchall()
|
|
assert rows[0][0] == 0
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_refresh_creates_table_even_without_an_api_key(monkeypatch):
|
|
monkeypatch.delenv("FMP_API_KEY", raising=False)
|
|
monkeypatch.delenv("Financial_Modeling_Prep_Api_Key", raising=False)
|
|
s = EventStore(":memory:")
|
|
try:
|
|
out = refresh_float_cache(s.con, ["AAA"], api_key=None)
|
|
assert out["no_key"] is True
|
|
rows = s.con.execute("SELECT count(*) FROM float_cache").fetchall()
|
|
assert rows[0][0] == 0
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_get_float_is_a_noop_on_a_fresh_schema():
|
|
s = EventStore(":memory:")
|
|
try:
|
|
assert get_float(s.con, "AAA") is None
|
|
finally:
|
|
s.close()
|