Hugging Face's logo

Spaces:
BlueMen
/
tradesenseAi
Configuration error

App Files Community
tradesenseAi / backend /tests /data /test_cache_integrity.py
Niranjan-Murugarasu024
feat: TradeSense AI Chairman Edition production release ready for Hugging Face Spaces
0125724
Raw History Blame Contribute Delete
6.71 kB
"""
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