feat: add daily market bar persistence
prakhar82 committed
Sep 13, 2026 at 19:17 UTC
c7125148cb5651d6f9695480b2f3f849a41903e3
7 files changed
+480
-4
ai/research-engine/app/models.py
+52
-2
@@ -1,12 +1,12 @@
1
from __future__ import annotations
2
3
-from datetime import datetime, timezone
3
+from datetime import date, datetime, timezone
4
from decimal import Decimal
5
from enum import StrEnum
6
from typing import Any
7
from uuid import UUID, uuid4
8
9
-from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, field_validator
9
+from pydantic import AnyHttpUrl, AwareDatetime, BaseModel, ConfigDict, Field, field_validator, model_validator
10
11
12
def _to_camel(value: str) -> str:
@@ -307,6 +307,56 @@ class MarketPriceObservation(ResearchBaseModel):
307
retrieved_at: datetime
308
309
310
+class DailyMarketBar(ResearchBaseModel):
311
+ """Public provider/day OHLCV evidence, separate from close-only history.
312
+
313
+ Decimal bounds match PostgreSQL NUMERIC(38,12); extra precision is rejected,
314
+ never silently rounded. Turnover is stored as supplied, without unit conversion.
315
+ Provider symbol is provenance, not canonical identity. Optional metrics may
316
+ remain missing, including all prices when only volume evidence is supplied.
317
+ """
318
+ global_instrument_id: UUID
319
+ trading_date: date
320
+ open: Decimal | None = Field(default=None, gt=0, max_digits=38, decimal_places=12, allow_inf_nan=False)
321
+ high: Decimal | None = Field(default=None, gt=0, max_digits=38, decimal_places=12, allow_inf_nan=False)
322
+ low: Decimal | None = Field(default=None, gt=0, max_digits=38, decimal_places=12, allow_inf_nan=False)
323
+ close: Decimal | None = Field(default=None, gt=0, max_digits=38, decimal_places=12, allow_inf_nan=False)
324
+ previous_close: Decimal | None = Field(default=None, gt=0, max_digits=38, decimal_places=12, allow_inf_nan=False)
325
+ volume: int | None = Field(default=None, strict=True, ge=0, le=9223372036854775807)
326
+ turnover: Decimal | None = Field(default=None, ge=0, max_digits=38, decimal_places=12, allow_inf_nan=False)
327
+ currency: str = Field(min_length=1, max_length=16)
328
+ provider: str = Field(min_length=1, max_length=120)
329
+ provider_symbol: str | None = Field(default=None, max_length=240)
330
+ source_mode: SourceMode
331
+ source_url: str = Field(min_length=1, max_length=1000)
332
+ retrieved_at: AwareDatetime
333
+
334
+ @field_validator("currency", "provider", "source_url", mode="before")
335
+ @classmethod
336
+ def strip_bar_metadata(cls, value):
337
+ return value.strip() if isinstance(value, str) else value
338
+
339
+ @field_validator("trading_date", mode="before")
340
+ @classmethod
341
+ def require_bar_date(cls, value):
342
+ if type(value) is date:
343
+ return value
344
+ if isinstance(value, str) and len(value) == 10:
345
+ return date.fromisoformat(value)
346
+ raise ValueError("tradingDate must be a DATE, not a timestamp")
347
+
348
+ @field_validator("retrieved_at")
349
+ @classmethod
350
+ def normalize_bar_retrieved_at(cls, value):
351
+ return value.astimezone(timezone.utc)
352
+
353
+ @model_validator(mode="after")
354
+ def validate_bar_range(self):
355
+ if self.high is not None and self.low is not None and self.high < self.low:
356
+ raise ValueError("daily bar high must be >= low")
357
+ return self
358
+
359
+
360
class PublicAnalyst(ResearchBaseModel):
361
target_low_price: Decimal | None = None
362
target_median_price: Decimal | None = None
ai/research-engine/app/persistence.py
+112
@@ -29,6 +29,7 @@ from app.models import (
29
StructuredMarketSnapshot,
30
StructuredMarketSnapshotRecord,
31
MarketPriceObservation,
32
+ DailyMarketBar,
33
)
34
from app.settings import Settings
35
from app.fact_precedence import FinancialFact, FinancialFactKey, FactSourceTier, merge_fact
@@ -106,6 +107,10 @@ class ResearchPersistence(Protocol):
107
def load_market_price_observations(self, instrument_ids: set[UUID] | None = None) -> list[MarketPriceObservation]: ...
108
def load_market_price_coverage(self, instrument_ids: set[UUID]) -> dict[UUID, tuple[datetime, datetime, int]]: ...
109
def upsert_market_price_observation(self, observation: MarketPriceObservation) -> None: ...
110
+ def upsert_daily_market_bar(self, bar: DailyMarketBar) -> None: ...
111
+ def upsert_daily_market_bars(self, bars: list[DailyMarketBar]) -> int: ...
112
+ def load_daily_market_bars(self, instrument_ids: set[UUID], *, start_date: date | None = None,
113
+ end_date: date | None = None, provider: str | None = None) -> list[DailyMarketBar]: ...
114
def record_structured_market_failure(self, instrument_id: UUID, provider: str, attempted_at: datetime, code: str, message: str) -> None: ...
115
def load_market_schedules(self, markets: set[str] | None = None): ...
116
def load_market_calendar_exceptions(self, markets: set[str] | None = None): ...
@@ -156,6 +161,9 @@ class DisabledResearchPersistence:
161
def load_market_price_observations(self, instrument_ids=None): return []
162
def load_market_price_coverage(self, instrument_ids): return {}
163
def upsert_market_price_observation(self, observation): return None
164
+ def upsert_daily_market_bar(self, bar): return None
165
+ def upsert_daily_market_bars(self, bars): return 0
166
+ def load_daily_market_bars(self, instrument_ids, *, start_date=None, end_date=None, provider=None): return []
167
def record_structured_market_failure(self, *args): return None
168
def load_market_schedules(self, markets=None): return []
169
def load_market_calendar_exceptions(self, markets=None): return []
@@ -186,6 +194,78 @@ class SqliteResearchPersistence:
194
self._connection.executescript(_sqlite_schema())
195
self._connection.commit()
196
197
+ def upsert_daily_market_bar(self, bar: DailyMarketBar) -> None:
198
+ self.upsert_daily_market_bars([bar])
199
+
200
+ def upsert_daily_market_bars(self, bars: list[DailyMarketBar]) -> int:
201
+ """Atomic correction of provider/day rows; last duplicate input wins.
202
+
203
+ Validate before writing (including model_copy/construct bypasses). Batch
204
+ 50 rows/750 parameters to stay below older SQLite's 999-parameter limit.
205
+ Returns the number of distinct identities supplied, not a change count.
206
+ """
207
+ selected = {}
208
+ for bar in bars:
209
+ validated = DailyMarketBar.model_validate(bar.model_dump())
210
+ key = (str(validated.global_instrument_id), validated.trading_date, validated.provider)
211
+ selected[key] = validated
212
+ ordered = [selected[key] for key in sorted(selected)]
213
+ if not ordered:
214
+ return 0
215
+ with self._connection:
216
+ for offset in range(0, len(ordered), 50):
217
+ batch = ordered[offset:offset + 50]
218
+ values = ",".join("(" + ",".join("?" for _ in range(15)) + ")" for _ in batch)
219
+ params = []
220
+ for bar in batch:
221
+ params.extend((str(bar.global_instrument_id), bar.trading_date.isoformat(),
222
+ _decimal(bar.open), _decimal(bar.high), _decimal(bar.low), _decimal(bar.close),
223
+ _decimal(bar.previous_close), bar.volume, _decimal(bar.turnover), bar.currency,
224
+ bar.provider, bar.provider_symbol, str(bar.source_mode), bar.source_url, _dt(bar.retrieved_at)))
225
+ self._connection.execute(f"""INSERT INTO global_daily_market_bars (
226
+ global_instrument_id,trading_date,open_price,high_price,low_price,close_price,
227
+ previous_close,volume,turnover,currency,provider,provider_symbol,source_mode,source_url,retrieved_at
228
+ ) VALUES {values}
229
+ ON CONFLICT(global_instrument_id,trading_date,provider) DO UPDATE SET
230
+ open_price=excluded.open_price,high_price=excluded.high_price,low_price=excluded.low_price,
231
+ close_price=excluded.close_price,previous_close=excluded.previous_close,volume=excluded.volume,
232
+ turnover=excluded.turnover,currency=excluded.currency,provider_symbol=excluded.provider_symbol,
233
+ source_mode=excluded.source_mode,source_url=excluded.source_url,retrieved_at=excluded.retrieved_at""", params)
234
+ return len(ordered)
235
+
236
+ def load_daily_market_bars(self, instrument_ids: set[UUID], *, start_date: date | None = None,
237
+ end_date: date | None = None, provider: str | None = None) -> list[DailyMarketBar]:
238
+ """Inclusive date bounds, explicit IDs only, stable global/date/provider order."""
239
+ if instrument_ids is None:
240
+ raise ValueError("Daily bars require an explicit instrument ID set")
241
+ if not instrument_ids:
242
+ return []
243
+ if any(value is not None and type(value) is not date for value in (start_date, end_date)):
244
+ raise ValueError("Daily bar range bounds must be DATE values")
245
+ if start_date is not None and end_date is not None and start_date > end_date:
246
+ raise ValueError("Daily bar start_date must be <= end_date")
247
+ if provider is not None and not provider.strip():
248
+ raise ValueError("Provider filter must be nonblank")
249
+ ordered = sorted({str(UUID(str(value))) for value in instrument_ids})
250
+ result = []
251
+ for offset in range(0, len(ordered), 500):
252
+ batch = ordered[offset:offset + 500]
253
+ params = list(batch)
254
+ sql = "SELECT * FROM global_daily_market_bars WHERE global_instrument_id IN (" + ",".join("?" for _ in batch) + ")"
255
+ if start_date is not None:
256
+ sql += " AND trading_date >= ?"
257
+ params.append(start_date.isoformat())
258
+ if end_date is not None:
259
+ sql += " AND trading_date <= ?"
260
+ params.append(end_date.isoformat())
261
+ if provider is not None:
262
+ sql += " AND provider = ?"
263
+ params.append(provider.strip())
264
+ sql += " ORDER BY global_instrument_id, trading_date, provider"
265
+ result.extend(_daily_market_bar_from_row(row) for row in self._connection.execute(sql, params).fetchall())
266
+ # Sort once more for identical ordering across database collations.
267
+ return sorted(result, key=lambda bar: (str(bar.global_instrument_id), bar.trading_date, bar.provider))
268
+
269
def _filtered_rows(self, table, ids, *, column="instrument_id", order=None):
270
"""Internal identifiers only; values are parameterized in bounded batches."""
271
if ids is not None:
@@ -763,6 +843,27 @@ def persistence_from_settings(settings: Settings) -> ResearchPersistence:
843
844
def _sqlite_schema() -> str:
845
return """
846
+ CREATE TABLE IF NOT EXISTS global_daily_market_bars (
847
+ global_instrument_id TEXT NOT NULL,
848
+ trading_date TEXT NOT NULL CHECK (length(trading_date) = 10 AND date(trading_date) IS NOT NULL AND date(trading_date) = trading_date),
849
+ open_price TEXT CHECK (CAST(open_price AS NUMERIC) > 0),
850
+ high_price TEXT CHECK (CAST(high_price AS NUMERIC) > 0),
851
+ low_price TEXT CHECK (CAST(low_price AS NUMERIC) > 0),
852
+ close_price TEXT CHECK (CAST(close_price AS NUMERIC) > 0),
853
+ previous_close TEXT CHECK (CAST(previous_close AS NUMERIC) > 0),
854
+ volume INTEGER CHECK (typeof(volume) = 'null' OR (typeof(volume) = 'integer' AND volume >= 0)),
855
+ turnover TEXT CHECK (CAST(turnover AS NUMERIC) >= 0),
856
+ currency TEXT NOT NULL CHECK (length(trim(currency)) > 0 AND length(currency) <= 16),
857
+ provider TEXT NOT NULL CHECK (length(trim(provider)) > 0 AND length(provider) <= 120),
858
+ provider_symbol TEXT CHECK (length(provider_symbol) <= 240),
859
+ source_mode TEXT NOT NULL CHECK (source_mode IN ('REAL', 'DEMO')),
860
+ source_url TEXT NOT NULL CHECK (length(trim(source_url)) > 0 AND length(source_url) <= 1000),
861
+ retrieved_at TEXT NOT NULL,
862
+ CONSTRAINT pk_global_daily_market_bars PRIMARY KEY (global_instrument_id, trading_date, provider),
863
+ CONSTRAINT ck_daily_bar_range CHECK (CAST(high_price AS NUMERIC) >= CAST(low_price AS NUMERIC))
864
+ );
865
+ CREATE INDEX IF NOT EXISTS idx_daily_market_bars_date_instrument
866
+ ON global_daily_market_bars (trading_date, global_instrument_id);
867
CREATE TABLE IF NOT EXISTS research_acquisition_observations (
868
instrument_id TEXT NOT NULL, requirement_id TEXT NOT NULL, provider TEXT NOT NULL,
869
outcome TEXT NOT NULL, observed_at TEXT NOT NULL, source_url TEXT,
@@ -983,6 +1084,17 @@ def _sqlite_schema() -> str:
1084
"""
1085
1086
1087
+def _daily_market_bar_from_row(row) -> DailyMarketBar:
1088
+ return DailyMarketBar(
1089
+ global_instrument_id=_required_uuid(row["global_instrument_id"], "global_daily_market_bars.global_instrument_id"),
1090
+ trading_date=row["trading_date"], open=_parse_decimal(row["open_price"]), high=_parse_decimal(row["high_price"]),
1091
+ low=_parse_decimal(row["low_price"]), close=_parse_decimal(row["close_price"]),
1092
+ previous_close=_parse_decimal(row["previous_close"]), volume=row["volume"], turnover=_parse_decimal(row["turnover"]),
1093
+ currency=row["currency"], provider=row["provider"], provider_symbol=row["provider_symbol"],
1094
+ source_mode=row["source_mode"], source_url=row["source_url"], retrieved_at=_parse_dt(row["retrieved_at"]),
1095
+ )
1096
+
1097
+
1098
def _document_from_row(row: sqlite3.Row) -> ResearchDocument:
1099
return ResearchDocument(
1100
document_id=_parse_uuid(row["document_id"]) or uuid4(),
ai/research-engine/app/postgres_persistence.py
+1
@@ -44,6 +44,7 @@ class PostgresResearchPersistence(SqliteResearchPersistence):
44
self._connection.execute("SELECT 1 FROM global_financial_facts LIMIT 0")
45
self._connection.execute("SELECT 1 FROM global_structured_market_snapshots LIMIT 0")
46
self._connection.execute("SELECT 1 FROM global_market_price_observations LIMIT 0")
47
+ self._connection.execute("SELECT 1 FROM global_daily_market_bars LIMIT 0")
48
self._connection.execute("SELECT 1 FROM global_stock_rule_engine_results LIMIT 0")
49
self._connection.execute("SELECT 1 FROM market_trading_schedules LIMIT 0")
50
self._connection.execute("SELECT 1 FROM market_trading_calendar_exceptions LIMIT 0")
ai/research-engine/app/repository.py
+21
-1
@@ -6,7 +6,7 @@ import logging
6
import threading
7
import time
8
from dataclasses import dataclass
9
-from datetime import datetime, timedelta, timezone
9
+from datetime import date, datetime, timedelta, timezone
10
from urllib.parse import urlparse
11
from uuid import UUID
12
@@ -33,6 +33,7 @@ from app.models import (
33
ShareholdingSnapshot,
34
SourceType,
35
StructuredMarketSnapshotRecord,
36
+ DailyMarketBar,
37
)
38
from app.normalization import canonicalize_url, content_hash, detect_document_type, extract_published_at, extract_text, normalize_text
39
from app.research_fetching import FetchError, HttpResearchFetcher, PdfExtractionTimeoutError, RestrictedFetchError, TransportFetchError
@@ -274,6 +275,25 @@ class ResearchRepository:
275
async def upsert_market_price_observation_async(self, observation) -> None:
276
await self._run_blocking_persistence(self._persistence.upsert_market_price_observation, observation)
277
278
+ def daily_market_bars_for(self, instrument_ids: set[UUID], *, start_date: date | None = None,
279
+ end_date: date | None = None, provider: str | None = None) -> dict[UUID, list[DailyMarketBar]]:
280
+ grouped = {instrument_id: [] for instrument_id in instrument_ids}
281
+ for bar in self._persistence.load_daily_market_bars(
282
+ instrument_ids, start_date=start_date, end_date=end_date, provider=provider):
283
+ grouped[bar.global_instrument_id].append(bar)
284
+ return grouped
285
+
286
+ async def daily_market_bars_for_instruments(self, instrument_ids: set[UUID], *, start_date: date | None = None,
287
+ end_date: date | None = None, provider: str | None = None) -> dict[UUID, list[DailyMarketBar]]:
288
+ return await self._run_blocking_persistence(self.daily_market_bars_for, instrument_ids,
289
+ start_date=start_date, end_date=end_date, provider=provider)
290
+
291
+ async def upsert_daily_market_bar_async(self, bar: DailyMarketBar) -> None:
292
+ await self._run_blocking_persistence(self._persistence.upsert_daily_market_bar, bar)
293
+
294
+ async def upsert_daily_market_bars_async(self, bars: list[DailyMarketBar]) -> int:
295
+ return await self._run_blocking_persistence(self._persistence.upsert_daily_market_bars, bars)
296
+
297
async def stock_rule_engine_result(
298
self,
299
global_instrument_id: UUID,
ai/research-engine/tests/test_daily_market_bars.py
new
+197
@@ -0,0 +1,197 @@
1
+from datetime import date, datetime, timedelta, timezone
2
+from decimal import Decimal
3
+import sqlite3
4
+from uuid import UUID
5
+
6
+import pytest
7
+from pydantic import ValidationError
8
+
9
+from app.models import DailyMarketBar, MarketPriceObservation
10
+from app.persistence import SqliteResearchPersistence, DisabledResearchPersistence, persistence_from_settings
11
+from app.postgres_persistence import PostgresResearchPersistence, _PostgresConnectionAdapter
12
+from app.repository import ResearchRepository
13
+from app.settings import Settings
14
+
15
+
16
+KEY = UUID(int=1)
17
+DAY = date(2026, 9, 1)
18
+NOW = datetime(2026, 9, 2, tzinfo=timezone.utc)
19
+
20
+
21
+def bar(**updates):
22
+ return DailyMarketBar.model_validate(dict(global_instrument_id=KEY, trading_date=DAY,
23
+ open="100.123456789012", high="105.123456789012", low="99.123456789012", close="103.123456789012",
24
+ previous_close=None, volume=None, turnover=None, currency="INR", provider="PUBLIC_SOURCE",
25
+ provider_symbol="ABC", source_mode="REAL", source_url="https://example.test/bars", retrieved_at=NOW) | updates)
26
+
27
+
28
+def test_fresh_sqlite_bootstrap_schema_indexes_and_repeat_startup(tmp_path):
29
+ path = tmp_path / "fresh.sqlite"
30
+ store = persistence_from_settings(Settings(research_persistence_enabled=True, research_database_backend="sqlite", research_database_name=str(path)))
31
+ columns = store._connection.execute("PRAGMA table_info(global_daily_market_bars)").fetchall()
32
+ assert {r['name'] for r in columns} == {"global_instrument_id", "trading_date", "open_price", "high_price", "low_price",
33
+ "close_price", "previous_close", "volume", "turnover", "currency", "provider", "provider_symbol", "source_mode", "source_url", "retrieved_at"}
34
+ assert [r['name'] for r in sorted(columns, key=lambda r:r['pk']) if r['pk']] == ["global_instrument_id", "trading_date", "provider"]
35
+ indexes = store._connection.execute("PRAGMA index_list(global_daily_market_bars)").fetchall()
36
+ assert any(r['unique'] for r in indexes)
37
+ assert any(r['name'] == 'idx_daily_market_bars_date_instrument' for r in indexes)
38
+ store.upsert_daily_market_bar(bar())
39
+ store.migrate()
40
+ assert SqliteResearchPersistence(path).load_daily_market_bars({KEY}) == [bar()]
41
+
42
+
43
+def test_existing_sqlite_database_receives_table_without_manual_ddl(tmp_path, monkeypatch):
44
+ import app.persistence as persistence
45
+ original = persistence._sqlite_schema
46
+ schema = original()
47
+ # Bootstrap an earlier SQLite schema using its normal mechanism, then reopen
48
+ # with the current bootstrap function. Existing close-only rows survive.
49
+ previous = schema[schema.index(' CREATE TABLE IF NOT EXISTS research_acquisition_observations'):]
50
+ path = tmp_path / 'upgrade.sqlite'
51
+ monkeypatch.setattr(persistence, '_sqlite_schema', lambda: previous)
52
+ old = SqliteResearchPersistence(path)
53
+ price = MarketPriceObservation(instrument_id=KEY, observed_at=NOW, retrieved_at=NOW, price=Decimal(100),
54
+ currency='INR', provider='EXISTING', source_url='https://example.test')
55
+ old.upsert_market_price_observation(price)
56
+ old._connection.close()
57
+ monkeypatch.setattr(persistence, '_sqlite_schema', original)
58
+ current = SqliteResearchPersistence(path)
59
+ assert current.load_daily_market_bars({KEY}) == []
60
+ assert current.load_market_price_observations({KEY}) == [price]
61
+
62
+
63
+def test_insert_round_trip_preserves_decimals_date_utc_and_nulls():
64
+ store = SqliteResearchPersistence()
65
+ original = bar(retrieved_at=NOW.astimezone(timezone(timedelta(hours=5, minutes=30))))
66
+ store.upsert_daily_market_bar(original)
67
+ loaded = store.load_daily_market_bars({KEY})[0]
68
+ assert loaded == original
69
+ assert loaded.open == Decimal('100.123456789012')
70
+ assert type(loaded.trading_date) is date and loaded.retrieved_at.utcoffset() == timedelta(0)
71
+ assert loaded.volume is loaded.turnover is loaded.previous_close is None
72
+ assert loaded.model_dump(by_alias=True)['globalInstrumentId'] == KEY
73
+ assert store.load_market_price_observations({KEY}) == []
74
+
75
+
76
+def test_batch_idempotence_correction_and_provider_symbol_not_identity():
77
+ store = SqliteResearchPersistence()
78
+ first, second = bar(), bar(provider='SECOND_SOURCE')
79
+ assert store.upsert_daily_market_bars([first, second]) == 2
80
+ assert store.upsert_daily_market_bars([first, second]) == 2
81
+ corrected = bar(open=101, high=111, low=98, close=108, previous_close=102, volume=0,
82
+ turnover=Decimal('1234567890123.123456789012'), provider_symbol='RENAMED', source_url='https://example.test/corrected',
83
+ source_mode='DEMO', retrieved_at=NOW+timedelta(days=1))
84
+ store.upsert_daily_market_bar(corrected)
85
+ assert store.load_daily_market_bars({KEY}) == [corrected, second]
86
+ assert store.load_daily_market_bars({KEY})[0].volume == 0
87
+ # A partial correction clears missing fields rather than preserving stale values.
88
+ store.upsert_daily_market_bar(bar(provider_symbol=None))
89
+ loaded = store.load_daily_market_bars({KEY})[0]
90
+ assert loaded.volume is loaded.turnover is loaded.previous_close is loaded.provider_symbol is None
91
+
92
+
93
+def test_large_batches_filter_order_and_no_n_plus_one():
94
+ store = SqliteResearchPersistence()
95
+ rows = [bar(global_instrument_id=UUID(int=i), provider=provider, trading_date=DAY+timedelta(days=day))
96
+ for i in range(1, 504) for provider in ('B', 'A') for day in (1, 0)]
97
+ queries = []
98
+ store._connection.set_trace_callback(queries.append)
99
+ store.upsert_daily_market_bars(list(reversed(rows)))
100
+ assert sum(q.startswith('INSERT INTO global_daily_market_bars') for q in queries) == (len(rows)+49)//50
101
+ queries.clear()
102
+ ids = {UUID(int=i) for i in range(1, 502)}
103
+ loaded = store.load_daily_market_bars(ids, start_date=DAY, end_date=DAY, provider='B')
104
+ assert [b.global_instrument_id.int for b in loaded] == list(range(1, 502))
105
+ assert all(b.trading_date == DAY and b.provider == 'B' for b in loaded)
106
+ assert len(queries) == 2 and all('global_instrument_id IN (' in q for q in queries)
107
+ all_rows = store.load_daily_market_bars({KEY, UUID(int=2)})
108
+ assert [(r.global_instrument_id.int,r.trading_date,r.provider) for r in all_rows] == sorted((r.global_instrument_id.int,r.trading_date,r.provider) for r in all_rows)
109
+
110
+
111
+def test_empty_ids_and_empty_batch_issue_no_queries():
112
+ store = SqliteResearchPersistence()
113
+ store.upsert_daily_market_bar(bar())
114
+ queries = []
115
+ store._connection.set_trace_callback(queries.append)
116
+ assert store.load_daily_market_bars(set()) == []
117
+ assert store.upsert_daily_market_bars([]) == 0
118
+ assert queries == []
119
+ with pytest.raises(ValueError): store.load_daily_market_bars(None)
120
+ with pytest.raises(ValueError): store.load_daily_market_bars({KEY}, start_date=DAY+timedelta(days=1), end_date=DAY)
121
+ with pytest.raises(ValueError): store.load_daily_market_bars({KEY}, provider=' ')
122
+ with pytest.raises(ValueError): store.load_daily_market_bars({KEY}, start_date=NOW)
123
+
124
+
125
+@pytest.mark.parametrize('updates', [
126
+ {'volume': -1}, {'volume': 1.5}, {'volume': True}, {'volume': 9223372036854775808},
127
+ {'high': 90, 'low': 100}, {'open': 0}, {'close': -1}, {'previous_close': 0}, {'turnover': -1},
128
+ {'currency': ' '}, {'provider': ' '}, {'trading_date': NOW}, {'trading_date': '2026-09-01T00:00:00Z'},
129
+ {'retrieved_at': NOW.replace(tzinfo=None)}, {'close': 'NaN'}, {'turnover': 'Infinity'},
130
+ {'close': '1.1234567890123'}, {'source_mode': 'UNKNOWN'}, {'global_instrument_id': None},
131
+])
132
+def test_invalid_domain_evidence_rejected(updates):
133
+ with pytest.raises(ValidationError): bar(**updates)
134
+
135
+
136
+def test_partial_rows_zero_turnover_and_bigint_volume_are_valid():
137
+ store = SqliteResearchPersistence()
138
+ value = bar(open=None, high=None, low=None, close=None, turnover=0, volume=9223372036854775807)
139
+ store.upsert_daily_market_bar(value)
140
+ assert store.load_daily_market_bars({KEY}) == [value]
141
+
142
+
143
+@pytest.mark.parametrize('column,value', [('volume', -1), ('high_price', '1'), ('close_price', '0'),
144
+ ('currency',' '), ('provider',' '), ('turnover','-1')])
145
+def test_database_constraints_reject_invalid_updates(column, value):
146
+ store = SqliteResearchPersistence()
147
+ store.upsert_daily_market_bar(bar())
148
+ with pytest.raises(sqlite3.IntegrityError):
149
+ store._connection.execute(f'UPDATE global_daily_market_bars SET {column}=?', (value,))
150
+ store._connection.rollback()
151
+ assert store.load_daily_market_bars({KEY}) == [bar()]
152
+
153
+
154
+def test_batch_validates_all_inputs_before_writing_and_last_correction_wins():
155
+ store = SqliteResearchPersistence()
156
+ invalid = bar().model_copy(update={'volume': -1})
157
+ with pytest.raises(ValidationError): store.upsert_daily_market_bars([bar(), invalid])
158
+ assert store.load_daily_market_bars({KEY}) == []
159
+ corrected = bar(close=104)
160
+ assert store.upsert_daily_market_bars([bar(), corrected]) == 1
161
+ assert store.load_daily_market_bars({KEY}) == [corrected]
162
+
163
+
164
+def test_postgres_uses_same_parameterized_batch_boundary():
165
+ queries = []
166
+ class Connection:
167
+ def execute(self, sql, params=None): queries.append((sql, params)); return self
168
+ def fetchall(self): return []
169
+ def commit(self): pass
170
+ def rollback(self): pass
171
+ store = PostgresResearchPersistence.__new__(PostgresResearchPersistence)
172
+ store._connection = _PostgresConnectionAdapter(Connection())
173
+ store.upsert_daily_market_bars([bar(global_instrument_id=UUID(int=i)) for i in range(1, 102)])
174
+ assert [len(params) for _,params in queries] == [750, 750, 15]
175
+ assert all('%s' in sql and '?' not in sql for sql,_ in queries)
176
+ queries.clear()
177
+ assert store.load_daily_market_bars({KEY}, start_date=DAY, end_date=DAY, provider="a' OR 1=1 --") == []
178
+ assert "a' OR 1=1 --" not in queries[0][0]
179
+ assert queries[0][1] == [str(KEY), DAY.isoformat(), DAY.isoformat(), "a' OR 1=1 --"]
180
+
181
+
182
+@pytest.mark.asyncio
183
+async def test_repository_boundary_and_disabled_mode_are_provider_free(monkeypatch):
184
+ import socket
185
+ monkeypatch.setattr(socket, 'create_connection', lambda *a,**k: pytest.fail('provider/network invoked'))
186
+ store = SqliteResearchPersistence()
187
+ repo = ResearchRepository.__new__(ResearchRepository)
188
+ repo._persistence = store
189
+ await repo.upsert_daily_market_bar_async(bar())
190
+ assert await repo.upsert_daily_market_bars_async([bar(provider='SECOND')]) == 1
191
+ assert await repo.daily_market_bars_for_instruments({KEY,UUID(int=2)}, start_date=DAY, end_date=DAY, provider='SECOND') == {
192
+ KEY: [bar(provider='SECOND')], UUID(int=2): []}
193
+ assert repo.daily_market_bars_for({KEY}) == {KEY: store.load_daily_market_bars({KEY})}
194
+ disabled = DisabledResearchPersistence()
195
+ assert disabled.load_daily_market_bars({KEY}) == []
196
+ assert disabled.upsert_daily_market_bars([bar()]) == 0
197
+ assert disabled.upsert_daily_market_bar(bar()) is None
services/research-service/src/main/resources/db/migration/V11__global_daily_market_bars.sql
new
+37
@@ -0,0 +1,37 @@
1
+-- Dedicated daily OHLCV evidence. No migration or dual-write from close-only prices.
2
+-- Provider symbols and retrieval times never participate in canonical identity.
3
+CREATE TABLE global_daily_market_bars (
4
+ global_instrument_id UUID NOT NULL,
5
+ trading_date DATE NOT NULL,
6
+ open_price NUMERIC(38, 12),
7
+ high_price NUMERIC(38, 12),
8
+ low_price NUMERIC(38, 12),
9
+ close_price NUMERIC(38, 12),
10
+ previous_close NUMERIC(38, 12),
11
+ volume BIGINT,
12
+ turnover NUMERIC(38, 12),
13
+ currency VARCHAR(16) NOT NULL,
14
+ provider VARCHAR(120) NOT NULL,
15
+ provider_symbol VARCHAR(240),
16
+ source_mode VARCHAR(16) NOT NULL,
17
+ source_url VARCHAR(1000) NOT NULL,
18
+ retrieved_at TIMESTAMP WITH TIME ZONE NOT NULL,
19
+ CONSTRAINT pk_global_daily_market_bars PRIMARY KEY (global_instrument_id, trading_date, provider),
20
+ CONSTRAINT ck_daily_bar_open CHECK (open_price > 0 AND open_price <= 99999999999999999999999999.999999999999),
21
+ CONSTRAINT ck_daily_bar_high CHECK (high_price > 0 AND high_price <= 99999999999999999999999999.999999999999),
22
+ CONSTRAINT ck_daily_bar_low CHECK (low_price > 0 AND low_price <= 99999999999999999999999999.999999999999),
23
+ CONSTRAINT ck_daily_bar_close CHECK (close_price > 0 AND close_price <= 99999999999999999999999999.999999999999),
24
+ CONSTRAINT ck_daily_bar_previous_close CHECK (previous_close > 0 AND previous_close <= 99999999999999999999999999.999999999999),
25
+ CONSTRAINT ck_daily_bar_range CHECK (high_price >= low_price),
26
+ CONSTRAINT ck_daily_bar_volume CHECK (volume >= 0),
27
+ CONSTRAINT ck_daily_bar_turnover CHECK (turnover >= 0 AND turnover <= 99999999999999999999999999.999999999999),
28
+ CONSTRAINT ck_daily_bar_currency CHECK (LENGTH(TRIM(currency)) > 0),
29
+ CONSTRAINT ck_daily_bar_provider CHECK (LENGTH(TRIM(provider)) > 0),
30
+ CONSTRAINT ck_daily_bar_source_mode CHECK (source_mode IN ('REAL', 'DEMO')),
31
+ CONSTRAINT ck_daily_bar_source_url CHECK (LENGTH(TRIM(source_url)) > 0)
32
+);
33
+
34
+-- The primary-key index covers instrument/date ranges and provider/day upserts.
35
+-- Date-first lookup covers recent bars across an instrument batch.
36
+CREATE INDEX idx_daily_market_bars_date_instrument
37
+ ON global_daily_market_bars (trading_date, global_instrument_id);
services/research-service/src/test/java/com/aiinvestment/research/ResearchFlywayMigrationTest.java
+60
-1
@@ -1,6 +1,7 @@
1
package com.aiinvestment.research;
2
3
import org.junit.jupiter.api.Test;
4
+import org.flywaydb.core.Flyway;
5
import org.springframework.beans.factory.annotation.Autowired;
6
import org.springframework.boot.test.context.SpringBootTest;
7
import org.springframework.jdbc.core.JdbcTemplate;
@@ -9,6 +10,7 @@ import org.springframework.test.context.ActiveProfiles;
10
import java.util.List;
11
12
import static org.assertj.core.api.Assertions.assertThat;
13
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
14
15
@SpringBootTest
16
@ActiveProfiles("test")
@@ -40,6 +42,7 @@ class ResearchFlywayMigrationTest {
42
"global_financial_facts",
43
"global_structured_market_snapshots",
44
"global_market_price_observations",
45
+ "global_daily_market_bars",
46
"global_stock_rule_engine_results",
47
"market_trading_schedules",
48
"market_trading_calendar_exceptions"
@@ -91,7 +94,7 @@ class ResearchFlywayMigrationTest {
94
""",
95
String.class
96
);
94
- assertThat(version).isEqualTo("10");
97
+ assertThat(version).isEqualTo("11");
98
99
Integer nseSessions = jdbcTemplate.queryForObject(
100
"SELECT count(*) FROM research.market_trading_schedules WHERE market_code = 'NSE'", Integer.class
@@ -107,5 +110,61 @@ class ResearchFlywayMigrationTest {
110
);
111
assertThat(exceptions).isZero();
112
}
113
+
114
+ @Test
115
+ void dailyBarBootstrapCreatesKeysIndexesDateTypesAndConstraints() throws Exception {
116
+ try (var connection = jdbcTemplate.getDataSource().getConnection()) {
117
+ var keys = new java.util.TreeMap<Short, String>();
118
+ try (var rows = connection.getMetaData().getPrimaryKeys(null, "research", "global_daily_market_bars")) {
119
+ while (rows.next()) keys.put(rows.getShort("KEY_SEQ"), rows.getString("COLUMN_NAME"));
120
+ }
121
+ assertThat(keys.values()).containsExactly("global_instrument_id", "trading_date", "provider");
122
+ var indexes = new java.util.HashSet<String>();
123
+ try (var rows = connection.getMetaData().getIndexInfo(null, "research", "global_daily_market_bars", false, false)) {
124
+ while (rows.next()) indexes.add(rows.getString("INDEX_NAME"));
125
+ }
126
+ assertThat(indexes).contains("idx_daily_market_bars_date_instrument");
127
+ try (var rows = connection.getMetaData().getColumns(null, "research", "global_daily_market_bars", "trading_date")) {
128
+ assertThat(rows.next()).isTrue();
129
+ assertThat(rows.getInt("DATA_TYPE")).isEqualTo(java.sql.Types.DATE);
130
+ }
131
+ }
132
+ String insert = """
133
+ INSERT INTO research.global_daily_market_bars
134
+ (global_instrument_id,trading_date,provider,currency,source_mode,source_url,retrieved_at,high_price,low_price,volume)
135
+ VALUES ('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',DATE '2026-09-01','TEST','INR','REAL','https://example.test',CURRENT_TIMESTAMP,?,?,?)
136
+ """;
137
+ assertThatThrownBy(() -> jdbcTemplate.update(insert, 10, 20, 1)).isInstanceOf(org.springframework.dao.DataAccessException.class);
138
+ assertThatThrownBy(() -> jdbcTemplate.update(insert, 20, 10, -1)).isInstanceOf(org.springframework.dao.DataAccessException.class);
139
+ jdbcTemplate.update(insert, 20, 10, 0);
140
+ assertThatThrownBy(() -> jdbcTemplate.update(insert, 20, 10, 0)).isInstanceOf(org.springframework.dao.DataAccessException.class);
141
+ jdbcTemplate.update("DELETE FROM research.global_daily_market_bars WHERE global_instrument_id = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'");
142
+ }
143
+
144
+ @Test
145
+ void existingVersionTenUpgradesAndRepeatedMigrationIsANoOp() {
146
+ // Optional disposable PostgreSQL database lets the same upgrade test run
147
+ // against the production dialect without changing application bootstrap.
148
+ String url = System.getProperty("dailyBarsUpgradeJdbcUrl",
149
+ "jdbc:h2:mem:dailyBarUpgrade;DB_CLOSE_DELAY=-1;MODE=PostgreSQL;DATABASE_TO_LOWER=TRUE");
150
+ String username = System.getProperty("dailyBarsUpgradeUsername", "sa");
151
+ String password = System.getProperty("dailyBarsUpgradePassword", "");
152
+ Flyway before = Flyway.configure().dataSource(url, username, password)
153
+ .schemas("research").defaultSchema("research").table("flyway_schema_history_research")
154
+ .target("10").load();
155
+ before.migrate();
156
+ JdbcTemplate existing = new JdbcTemplate(new org.springframework.jdbc.datasource.DriverManagerDataSource(url, username, password));
157
+ existing.update("""
158
+ INSERT INTO research.global_market_price_observations
159
+ (instrument_id,observed_at,price,provider,source_url,retrieved_at)
160
+ VALUES ('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',CURRENT_TIMESTAMP,100,'EXISTING','https://example.test',CURRENT_TIMESTAMP)
161
+ """);
162
+ Flyway upgrade = Flyway.configure().dataSource(url, username, password)
163
+ .schemas("research").defaultSchema("research").table("flyway_schema_history_research").load();
164
+ assertThat(upgrade.migrate().migrationsExecuted).isEqualTo(1);
165
+ assertThat(upgrade.migrate().migrationsExecuted).isZero();
166
+ assertThat(existing.queryForObject("SELECT COUNT(*) FROM research.global_daily_market_bars", Integer.class)).isZero();
167
+ assertThat(existing.queryForObject("SELECT COUNT(*) FROM research.global_market_price_observations", Integer.class)).isEqualTo(1);
168
+ }
169
}
170