Spaces:
Configuration error
Configuration error
File size: 6,712 Bytes
0125724 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | """
TradeSense AI — Market Data Cache Integrity Tests
Tests cache hit/miss semantics, TTL expiry, temporal cutoff validation on retrieval,
and confirms that API keys/secrets are NEVER cached in metadata.
"""
import time
from datetime import date, datetime, timezone
import pandas as pd
import pytest
from backend.config.settings import settings
from backend.app.schemas.contracts import (
DatasetMetadata,
DataQualityReport,
DataQualityStatus,
DataFreshness,
CrossSourceAgreement,
AdjustmentMethod,
)
from backend.data.cache import MarketDataCache, CacheEntryMetadata
def create_sample_df(ticker: str = "AAPL", days: int = 10, end_date: date = date(2024, 12, 31)) -> pd.DataFrame:
dates = pd.bdate_range(end=end_date, periods=days).date
return pd.DataFrame({
"session": dates,
"Open": [150.0 + i for i in range(days)],
"High": [155.0 + i for i in range(days)],
"Low": [149.0 + i for i in range(days)],
"Close": [153.0 + i for i in range(days)],
"Volume": [1000000.0 for _ in range(days)],
"VWAP": [152.0 + i for i in range(days)],
})
def create_sample_metadata(ticker: str = "AAPL", cutoff: date = date(2024, 12, 31), count: int = 10) -> DatasetMetadata:
return DatasetMetadata(
dataset_version=f"ds_{ticker.lower()}_20241231_abc12345",
snapshot_id=f"snap_{ticker.lower()}_123",
ticker=ticker,
source_primary="Polygon.io",
source_secondary=None,
fallback_used=False,
retrieved_at=datetime.now(timezone.utc),
start_session=date(2024, 1, 2),
cutoff_session=cutoff,
total_bars=count,
splits_adjusted=True,
dividends_adjusted=True,
adjustment_method=AdjustmentMethod.PROVIDER_ADJUSTED,
timezone="America/New_York",
quality_report=DataQualityReport(
quality_status=DataQualityStatus.VERIFIED,
quality_score=95,
source="Polygon.io",
fallback_used=False,
cross_source_agreement=CrossSourceAgreement.UNCHECKED,
freshness=DataFreshness.LATEST_AVAILABLE,
corporate_action_status="VALIDATED",
calendar_status="VALIDATED",
temporal_integrity="PASS",
ohlc_invariants_passed=True,
),
sha256_hash="e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
)
class TestCacheIntegrity:
def setup_method(self):
self.cache = MarketDataCache()
def test_cache_miss_on_empty_cache(self):
result = self.cache.get("AAPL", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31))
assert result is None
def test_cache_set_and_get_hit(self):
df = create_sample_df("AAPL")
meta = create_sample_metadata("AAPL")
self.cache.set("AAPL", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31), df, meta)
hit = self.cache.get("AAPL", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31))
assert hit is not None
cached_df, cached_meta = hit
assert len(cached_df) == len(df)
assert cached_meta.ticker == "AAPL"
assert cached_meta.sha256_hash == meta.sha256_hash
def test_cache_miss_on_different_ticker(self):
df = create_sample_df("AAPL")
meta = create_sample_metadata("AAPL")
self.cache.set("AAPL", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31), df, meta)
hit = self.cache.get("MSFT", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31))
assert hit is None
def test_cache_temporal_cutoff_enforcement(self):
"""A cache entry stored for cutoff T1 must NOT match a request for cutoff T2."""
df = create_sample_df("AAPL", end_date=date(2024, 6, 30))
meta = create_sample_metadata("AAPL", cutoff=date(2024, 6, 30))
self.cache.set("AAPL", date(2024, 1, 1), date(2024, 6, 30), date(2024, 6, 30), df, meta)
# Requesting with different cutoff date must miss
hit = self.cache.get("AAPL", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31))
assert hit is None
def test_cache_expiry_enforcement(self):
"""Expired entries must return None and be evicted."""
df = create_sample_df("NVDA")
meta = create_sample_metadata("NVDA")
# Store with 0-second TTL (instant expiration)
key = self.cache._build_key("NVDA", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31))
from backend.data.cache import CacheEntry
self.cache._cache[key] = CacheEntry(df, meta, ttl_seconds=-1)
hit = self.cache.get("NVDA", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31))
assert hit is None
# Verify it was evicted from internal dict
assert key not in self.cache._cache
def test_cache_metadata_never_contains_credentials(self):
"""Safe dict export must contain only public audit fields and never secrets."""
df = create_sample_df("TSLA")
meta = create_sample_metadata("TSLA")
self.cache.set("TSLA", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31), df, meta)
entry_meta = self.cache.get_entry_metadata(
"TSLA", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31)
)
assert entry_meta is not None
assert "ticker" in entry_meta
assert "lineage_hash" in entry_meta
assert "provider" in entry_meta
# Check no sensitive keys
forbidden = ["api_key", "apiKey", "secret", "token", "password", "authorization"]
for key in entry_meta:
assert key.lower() not in forbidden
val_str = str(entry_meta[key]).lower()
for f in forbidden:
assert f not in val_str
def test_cache_invalidation_by_ticker(self):
df = create_sample_df("AMZN")
meta = create_sample_metadata("AMZN")
self.cache.set("AMZN", date(2024, 1, 1), date(2024, 6, 30), date(2024, 6, 30), df, meta)
self.cache.set("AMZN", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31), df, meta)
evicted = self.cache.invalidate("AMZN")
assert evicted == 2
assert self.cache.get("AMZN", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31)) is None
def test_cache_stats_accuracy(self):
self.cache.clear()
df = create_sample_df("SPY")
meta = create_sample_metadata("SPY")
self.cache.set("SPY", date(2024, 1, 1), date(2024, 12, 31), date(2024, 12, 31), df, meta)
stats = self.cache.stats()
assert stats["total_entries"] == 1
assert stats["live_entries"] == 1
assert stats["expired_entries"] == 0
|