diff --git a/app/pt_backup.py b/app/pt_backup.py new file mode 100644 index 00000000..184d2a82 --- /dev/null +++ b/app/pt_backup.py @@ -0,0 +1,421 @@ +""" +PowerTraderAI+ Automated Database Backup & Restore System +Handles scheduled SQLite backups, integrity verification, and +point-in-time recovery with configurable retention policy. +""" + +import hashlib +import json +import logging +import os +import shutil +import sqlite3 +import threading +import time +from dataclasses import dataclass, asdict +from datetime import datetime +from typing import Dict, List, Optional + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Data classes +# --------------------------------------------------------------------------- +@dataclass +class BackupRecord: + """Metadata for a single backup file.""" + + backup_id: str + source_db: str + backup_path: str + created_at: float + file_size_bytes: int + sha256_checksum: str + integrity_check_passed: bool + notes: str = "" + + def created_at_iso(self) -> str: + return datetime.fromtimestamp(self.created_at).isoformat() + + def to_dict(self) -> dict: + d = asdict(self) + d["created_at_iso"] = self.created_at_iso() + return d + + @classmethod + def from_dict(cls, d: dict) -> "BackupRecord": + d.pop("created_at_iso", None) + return cls(**{k: v for k, v in d.items() if k in cls.__dataclass_fields__}) + + +@dataclass +class RestoreResult: + """Result of a restore operation.""" + + success: bool + backup_id: str + restored_from: str + restored_to: str + message: str + timestamp: float + + +# --------------------------------------------------------------------------- +# DatabaseBackupManager +# --------------------------------------------------------------------------- +class DatabaseBackupManager: + """ + Manages automated backup and restore for SQLite databases. + + Features: + - SHA-256 checksums for backup integrity verification + - SQLite PRAGMA integrity_check before backup + - WAL checkpoint to ensure all data is flushed + - Configurable retention (keep last N backups) + - Point-in-time recovery (restore any stored backup) + - Background scheduler thread for automated backups + + Args: + db_path: Path to the SQLite database file + backup_dir: Directory to store backup files (auto-created) + max_backups: Maximum number of backups to retain + backup_interval_hours: Hours between automated backups + """ + + MANIFEST_FILE = "backup_manifest.json" + + def __init__( + self, + db_path: str, + backup_dir: Optional[str] = None, + max_backups: int = 30, + backup_interval_hours: float = 24.0, + ): + self.db_path = os.path.abspath(db_path) + self.backup_dir = backup_dir or os.path.join( + os.path.dirname(self.db_path), "backups" + ) + self.max_backups = max_backups + self.backup_interval = backup_interval_hours * 3600 + self._manifest_path = os.path.join(self.backup_dir, self.MANIFEST_FILE) + self._lock = threading.RLock() + self._stop_event = threading.Event() + self._scheduler_thread: Optional[threading.Thread] = None + os.makedirs(self.backup_dir, exist_ok=True) + + # ------------------------------------------------------------------ + # Checksum + # ------------------------------------------------------------------ + def _sha256(self, path: str) -> str: + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + return h.hexdigest() + + # ------------------------------------------------------------------ + # Integrity check + # ------------------------------------------------------------------ + def check_db_integrity(self, db_path: Optional[str] = None) -> bool: + """Run SQLite integrity_check PRAGMA. Returns True if OK.""" + path = db_path or self.db_path + if not os.path.exists(path): + logger.warning("DB integrity check skipped — file not found: %s", path) + return False + try: + conn = sqlite3.connect(path) + cursor = conn.execute("PRAGMA integrity_check") + result = cursor.fetchone() + conn.close() + ok = result and result[0] == "ok" + if not ok: + logger.error("DB integrity check FAILED for %s: %s", path, result) + return ok + except sqlite3.Error as exc: + logger.error("DB integrity check error for %s: %s", path, exc) + return False + + # ------------------------------------------------------------------ + # Backup + # ------------------------------------------------------------------ + def create_backup(self, notes: str = "") -> Optional[BackupRecord]: + """ + Create a verified backup of the database. + + Steps: + 1. WAL checkpoint (flush WAL to main DB) + 2. Integrity check source DB + 3. Copy to backup file + 4. Verify backup checksum + 5. Integrity check backup + 6. Register in manifest + 7. Prune old backups if over limit + + Returns: + BackupRecord on success, None on failure. + """ + with self._lock: + if not os.path.exists(self.db_path): + logger.error("Source DB not found: %s", self.db_path) + return None + + # Step 1: WAL checkpoint + try: + conn = sqlite3.connect(self.db_path) + conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + conn.close() + except sqlite3.Error as exc: + logger.warning("WAL checkpoint failed (non-fatal): %s", exc) + + # Step 2: Integrity check + if not self.check_db_integrity(): + logger.error("Source DB integrity check failed — aborting backup") + return None + + # Step 3: Copy + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + backup_id = f"backup_{ts}" + backup_path = os.path.join(self.backup_dir, f"{backup_id}.db") + + try: + shutil.copy2(self.db_path, backup_path) + except OSError as exc: + logger.error("Failed to copy DB to backup: %s", exc) + return None + + # Step 4 & 5: Checksum + integrity on backup + checksum = self._sha256(backup_path) + integrity_ok = self.check_db_integrity(backup_path) + file_size = os.path.getsize(backup_path) + + record = BackupRecord( + backup_id=backup_id, + source_db=self.db_path, + backup_path=backup_path, + created_at=time.time(), + file_size_bytes=file_size, + sha256_checksum=checksum, + integrity_check_passed=integrity_ok, + notes=notes, + ) + + # Step 6: Register + self._append_manifest(record) + logger.info( + "Backup created: %s (%d bytes, integrity=%s)", + backup_id, + file_size, + integrity_ok, + ) + + # Step 7: Prune + self._prune_old_backups() + + return record + + # ------------------------------------------------------------------ + # Verify + # ------------------------------------------------------------------ + def verify_backup(self, backup_id: str) -> bool: + """Verify a backup's SHA-256 checksum and SQLite integrity.""" + record = self._get_record(backup_id) + if not record: + logger.error("Backup record not found: %s", backup_id) + return False + if not os.path.exists(record.backup_path): + logger.error("Backup file missing: %s", record.backup_path) + return False + + current_checksum = self._sha256(record.backup_path) + if current_checksum != record.sha256_checksum: + logger.error( + "Backup checksum MISMATCH for %s: expected %s, got %s", + backup_id, + record.sha256_checksum, + current_checksum, + ) + return False + + integrity = self.check_db_integrity(record.backup_path) + logger.info( + "Backup %s verified: checksum OK, integrity=%s", backup_id, integrity + ) + return integrity + + # ------------------------------------------------------------------ + # Restore + # ------------------------------------------------------------------ + def restore( + self, backup_id: str, target_path: Optional[str] = None + ) -> RestoreResult: + """ + Restore a backup to target_path (default: overwrites source DB). + + Safety: atomic swap — copy to temp file, verify, then rename. + """ + with self._lock: + record = self._get_record(backup_id) + if not record: + return RestoreResult( + success=False, + backup_id=backup_id, + restored_from="", + restored_to="", + message=f"Backup record not found: {backup_id}", + timestamp=time.time(), + ) + + # Verify before restore + if not self.verify_backup(backup_id): + return RestoreResult( + success=False, + backup_id=backup_id, + restored_from=record.backup_path, + restored_to="", + message="Backup verification failed — restore aborted", + timestamp=time.time(), + ) + + target = target_path or self.db_path + tmp_path = target + ".restore_tmp" + + try: + shutil.copy2(record.backup_path, tmp_path) + # Atomic replace + os.replace(tmp_path, target) + logger.info("Restored %s → %s", backup_id, target) + return RestoreResult( + success=True, + backup_id=backup_id, + restored_from=record.backup_path, + restored_to=target, + message="Restore successful", + timestamp=time.time(), + ) + except OSError as exc: + try: + os.remove(tmp_path) + except OSError: + pass + logger.error("Restore failed: %s", exc) + return RestoreResult( + success=False, + backup_id=backup_id, + restored_from=record.backup_path, + restored_to=target, + message=f"Restore failed: {exc}", + timestamp=time.time(), + ) + + def restore_latest(self, target_path: Optional[str] = None) -> RestoreResult: + """Restore the most recent backup.""" + records = self.list_backups() + if not records: + return RestoreResult( + success=False, + backup_id="", + restored_from="", + restored_to="", + message="No backups available", + timestamp=time.time(), + ) + return self.restore(records[-1].backup_id, target_path) + + # ------------------------------------------------------------------ + # List / manifest + # ------------------------------------------------------------------ + def list_backups(self) -> List[BackupRecord]: + """Return all backup records, sorted oldest-first.""" + manifest = self._load_manifest() + return sorted(manifest, key=lambda r: r.created_at) + + def _load_manifest(self) -> List[BackupRecord]: + if not os.path.exists(self._manifest_path): + return [] + try: + with open(self._manifest_path, "r", encoding="utf-8") as f: + entries = json.load(f) + return [BackupRecord.from_dict(e) for e in entries] + except (OSError, json.JSONDecodeError, TypeError): + return [] + + def _save_manifest(self, records: List[BackupRecord]) -> None: + with open(self._manifest_path, "w", encoding="utf-8") as f: + json.dump([r.to_dict() for r in records], f, indent=2) + + def _append_manifest(self, record: BackupRecord) -> None: + records = self._load_manifest() + records.append(record) + self._save_manifest(records) + + def _get_record(self, backup_id: str) -> Optional[BackupRecord]: + return next( + (r for r in self._load_manifest() if r.backup_id == backup_id), None + ) + + def _prune_old_backups(self) -> int: + """Delete oldest backups when over max_backups limit. Returns count pruned.""" + records = self.list_backups() + to_prune = records[: max(0, len(records) - self.max_backups)] + for record in to_prune: + try: + os.remove(record.backup_path) + logger.info("Pruned old backup: %s", record.backup_id) + except OSError as exc: + logger.warning("Could not remove old backup file: %s", exc) + if to_prune: + remaining = [r for r in records if r not in to_prune] + self._save_manifest(remaining) + return len(to_prune) + + # ------------------------------------------------------------------ + # Scheduler + # ------------------------------------------------------------------ + def start_scheduler(self) -> None: + """Start automated backup scheduler as daemon thread.""" + if self._scheduler_thread and self._scheduler_thread.is_alive(): + return + self._stop_event.clear() + self._scheduler_thread = threading.Thread( + target=self._run_scheduler, + name="DBBackupScheduler", + daemon=True, + ) + self._scheduler_thread.start() + logger.info( + "DB backup scheduler started (interval: %gh, max_backups: %d)", + self.backup_interval / 3600, + self.max_backups, + ) + + def stop_scheduler(self) -> None: + self._stop_event.set() + if self._scheduler_thread: + self._scheduler_thread.join(timeout=5) + logger.info("DB backup scheduler stopped") + + def _run_scheduler(self) -> None: + while not self._stop_event.is_set(): + try: + record = self.create_backup(notes="scheduled") + if record: + logger.info("Scheduled backup complete: %s", record.backup_id) + else: + logger.error("Scheduled backup FAILED") + except Exception as exc: + logger.error("Backup scheduler error: %s", exc) + self._stop_event.wait(timeout=self.backup_interval) + + def get_status(self) -> Dict: + """Return backup system status summary.""" + records = self.list_backups() + return { + "total_backups": len(records), + "latest_backup": records[-1].to_dict() if records else None, + "backup_dir": self.backup_dir, + "max_backups": self.max_backups, + "scheduler_running": bool( + self._scheduler_thread and self._scheduler_thread.is_alive() + ), + } diff --git a/app/pt_validation.py b/app/pt_validation.py index b7cecc3b..1874b99a 100644 --- a/app/pt_validation.py +++ b/app/pt_validation.py @@ -1,12 +1,15 @@ """ Input validation and sanitization for PowerTraderAI+. -Provides comprehensive validation for external data sources and user inputs. +Provides comprehensive validation for external data sources, user inputs, +and data corruption/anomaly detection. """ +import hashlib import json +import math import re from decimal import Decimal, InvalidOperation -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Optional, Tuple, Union class ValidationError(Exception): @@ -15,6 +18,12 @@ class ValidationError(Exception): pass +class DataCorruptionError(Exception): + """Raised when data integrity/corruption is detected.""" + + pass + + class InputValidator: """Comprehensive input validation for trading data.""" @@ -350,6 +359,189 @@ def validate_config_data(config_data: Dict[str, Any]) -> Dict[str, Any]: return validated +# --------------------------------------------------------------------------- +# DataIntegrityValidator — corruption detection (Issue #63) +# --------------------------------------------------------------------------- +class DataIntegrityValidator: + """ + Detects data corruption and anomalies in trading and market data. + + Checks: + - NaN / Infinity values in numeric fields + - OHLCV cross-field consistency (high >= low, close within range, etc.) + - Price spike detection (configurable Z-score threshold) + - Missing required fields + - SHA-256 checksum verification for serialized data blobs + """ + + # Default anomaly thresholds + PRICE_SPIKE_Z_SCORE = 5.0 # Flag if value deviates > 5 std devs from series mean + MAX_MISSING_FIELD_RATIO = 0.1 # Allow at most 10% missing fields in a batch + + @staticmethod + def has_nan_or_inf(value: Any) -> bool: + """True if value is float NaN or +/-Infinity.""" + if isinstance(value, float): + return math.isnan(value) or math.isinf(value) + if isinstance(value, Decimal): + return not value.is_finite() + return False + + @staticmethod + def check_ohlcv_consistency(candle: Dict[str, Any]) -> List[str]: + """ + Check OHLCV candle for internal consistency. + Returns list of violation messages (empty = clean). + """ + violations = [] + fields = {} + for f in ("open", "high", "low", "close"): + if f in candle: + try: + fields[f] = float(candle[f]) + if DataIntegrityValidator.has_nan_or_inf(fields[f]): + violations.append(f"NaN/Inf in field '{f}'") + except (ValueError, TypeError): + violations.append(f"Non-numeric value in field '{f}'") + + if len(fields) < 4: + return violations # Can't do cross-checks with missing fields + + high, low, open_, close = ( + fields.get("high"), + fields.get("low"), + fields.get("open"), + fields.get("close"), + ) + + if high is not None and low is not None and high < low: + violations.append(f"high ({high}) < low ({low})") + if high is not None and open_ is not None and open_ > high: + violations.append(f"open ({open_}) > high ({high})") + if ( + high is not None and close is not None and close > high * 1.001 + ): # 0.1% tolerance + violations.append(f"close ({close}) > high ({high})") + if low is not None and open_ is not None and open_ < low * 0.999: + violations.append(f"open ({open_}) < low ({low})") + if low is not None and close is not None and close < low * 0.999: + violations.append(f"close ({close}) < low ({low})") + + # Volume check + if "volume" in candle: + try: + vol = float(candle["volume"]) + if vol < 0: + violations.append(f"Negative volume: {vol}") + if DataIntegrityValidator.has_nan_or_inf(vol): + violations.append("NaN/Inf volume") + except (ValueError, TypeError): + violations.append("Non-numeric volume") + + return violations + + @staticmethod + def detect_price_spikes( + prices: List[float], + z_threshold: float = PRICE_SPIKE_Z_SCORE, + ) -> List[Tuple[int, float]]: + """ + Detect price spikes using Z-score. + Returns list of (index, value) tuples for anomalous entries. + """ + if len(prices) < 3: + return [] + import statistics + + try: + mean = statistics.mean(prices) + std = statistics.stdev(prices) + except statistics.StatisticsError: + return [] + if std == 0: + return [] + return [ + (i, v) for i, v in enumerate(prices) if abs((v - mean) / std) > z_threshold + ] + + @staticmethod + def validate_numeric_fields( + data: Dict[str, Any], numeric_fields: List[str] + ) -> List[str]: + """ + Validate that specified fields are finite numeric values. + Returns list of violation messages. + """ + violations = [] + for field in numeric_fields: + if field not in data: + violations.append(f"Missing field: '{field}'") + continue + val = data[field] + if val is None: + violations.append(f"Null value in field '{field}'") + continue + try: + fval = float(val) + if DataIntegrityValidator.has_nan_or_inf(fval): + violations.append(f"NaN/Inf in field '{field}': {val}") + except (ValueError, TypeError): + violations.append(f"Non-numeric value in field '{field}': {val!r}") + return violations + + @staticmethod + def compute_checksum(data: Union[str, bytes, Dict]) -> str: + """Compute SHA-256 checksum of data (for integrity tracking).""" + if isinstance(data, dict): + raw = json.dumps(data, sort_keys=True, default=str).encode("utf-8") + elif isinstance(data, str): + raw = data.encode("utf-8") + else: + raw = data + return hashlib.sha256(raw).hexdigest() + + @staticmethod + def verify_checksum(data: Union[str, bytes, Dict], expected: str) -> bool: + """Verify data matches expected checksum.""" + return DataIntegrityValidator.compute_checksum(data) == expected + + @staticmethod + def check_batch_integrity( + records: List[Dict[str, Any]], + required_fields: List[str], + ) -> Dict[str, Any]: + """ + Check a batch of records for integrity issues. + Returns summary: total, clean, corrupt, missing_field_count, anomalies. + """ + total = len(records) + corrupt_indices = [] + issues: Dict[int, List[str]] = {} + + for i, record in enumerate(records): + record_issues = [] + # NaN/Inf scan + for k, v in record.items(): + if DataIntegrityValidator.has_nan_or_inf(v): + record_issues.append(f"NaN/Inf in '{k}'") + # Missing fields + for f in required_fields: + if f not in record or record[f] is None: + record_issues.append(f"Missing/null '{f}'") + if record_issues: + corrupt_indices.append(i) + issues[i] = record_issues + + return { + "total": total, + "clean": total - len(corrupt_indices), + "corrupt": len(corrupt_indices), + "corrupt_indices": corrupt_indices, + "issues": issues, + "integrity_ok": len(corrupt_indices) == 0, + } + + def safe_json_loads(json_str: str, max_size: int = 1024 * 1024) -> Dict[str, Any]: """Safely parse JSON with size limits and error handling.""" if not isinstance(json_str, str): diff --git a/app/test_backup_validation.py b/app/test_backup_validation.py new file mode 100644 index 00000000..f4739b8b --- /dev/null +++ b/app/test_backup_validation.py @@ -0,0 +1,252 @@ +"""Tests for pt_backup (issue #62) and DataIntegrityValidator (issue #63).""" + +import os +import sqlite3 +import tempfile +import time +import unittest + +from pt_backup import DatabaseBackupManager +from pt_validation import DataIntegrityValidator + + +# --------------------------------------------------------------------------- +# pt_backup tests +# --------------------------------------------------------------------------- +def _make_test_db(path: str) -> None: + """Create a minimal SQLite DB for testing.""" + conn = sqlite3.connect(path) + conn.execute( + "CREATE TABLE orders (id INTEGER PRIMARY KEY, symbol TEXT, amount REAL)" + ) + conn.execute("INSERT INTO orders VALUES (1, 'BTC-USD', 0.5)") + conn.commit() + conn.close() + + +class TestDatabaseBackupManager(unittest.TestCase): + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + self.db_path = os.path.join(self.tmpdir, "test.db") + self.backup_dir = os.path.join(self.tmpdir, "backups") + _make_test_db(self.db_path) + self.mgr = DatabaseBackupManager( + db_path=self.db_path, + backup_dir=self.backup_dir, + max_backups=5, + ) + + def test_creates_backup(self): + record = self.mgr.create_backup() + self.assertIsNotNone(record) + self.assertTrue(os.path.exists(record.backup_path)) + + def test_backup_has_checksum(self): + record = self.mgr.create_backup() + self.assertTrue(len(record.sha256_checksum) == 64) + + def test_backup_integrity_check_passes(self): + record = self.mgr.create_backup() + self.assertTrue(record.integrity_check_passed) + + def test_verify_backup_clean(self): + record = self.mgr.create_backup() + self.assertTrue(self.mgr.verify_backup(record.backup_id)) + + def test_verify_backup_detects_tamper(self): + record = self.mgr.create_backup() + # Corrupt backup file + with open(record.backup_path, "ab") as f: + f.write(b"CORRUPTED") + self.assertFalse(self.mgr.verify_backup(record.backup_id)) + + def test_verify_missing_backup(self): + self.assertFalse(self.mgr.verify_backup("nonexistent_backup")) + + def test_list_backups(self): + self.mgr.create_backup() + self.mgr.create_backup() + records = self.mgr.list_backups() + self.assertEqual(len(records), 2) + + def test_list_backups_sorted_oldest_first(self): + self.mgr.create_backup() + time.sleep(0.01) + self.mgr.create_backup() + records = self.mgr.list_backups() + self.assertLessEqual(records[0].created_at, records[1].created_at) + + def test_restore_success(self): + record = self.mgr.create_backup() + restore_path = os.path.join(self.tmpdir, "restored.db") + result = self.mgr.restore(record.backup_id, target_path=restore_path) + self.assertTrue(result.success) + self.assertTrue(os.path.exists(restore_path)) + + def test_restore_data_intact(self): + record = self.mgr.create_backup() + restore_path = os.path.join(self.tmpdir, "restored2.db") + self.mgr.restore(record.backup_id, target_path=restore_path) + conn = sqlite3.connect(restore_path) + row = conn.execute("SELECT amount FROM orders WHERE id=1").fetchone() + conn.close() + self.assertAlmostEqual(row[0], 0.5) + + def test_restore_latest(self): + self.mgr.create_backup() + time.sleep(0.01) + self.mgr.create_backup() + restore_path = os.path.join(self.tmpdir, "restored_latest.db") + result = self.mgr.restore_latest(target_path=restore_path) + self.assertTrue(result.success) + + def test_restore_nonexistent_returns_failure(self): + result = self.mgr.restore("no_such_backup") + self.assertFalse(result.success) + + def test_prune_keeps_max_backups(self): + mgr = DatabaseBackupManager(self.db_path, self.backup_dir, max_backups=3) + for _ in range(5): + mgr.create_backup() + self.assertLessEqual(len(mgr.list_backups()), 3) + + def test_db_integrity_check(self): + self.assertTrue(self.mgr.check_db_integrity()) + + def test_db_integrity_check_missing_file(self): + self.assertFalse(self.mgr.check_db_integrity("/nonexistent/path.db")) + + def test_get_status(self): + self.mgr.create_backup() + status = self.mgr.get_status() + self.assertEqual(status["total_backups"], 1) + self.assertIsNotNone(status["latest_backup"]) + + def test_scheduler_start_stop(self): + self.mgr.start_scheduler() + self.assertTrue(self.mgr._scheduler_thread.is_alive()) + self.mgr.stop_scheduler() + self.assertFalse(self.mgr._scheduler_thread.is_alive()) + + def test_backup_manifest_persists(self): + self.mgr.create_backup() + # Re-create manager to re-read manifest from disk + mgr2 = DatabaseBackupManager(self.db_path, self.backup_dir) + self.assertEqual(len(mgr2.list_backups()), 1) + + def tearDown(self): + import shutil + + shutil.rmtree(self.tmpdir, ignore_errors=True) + + +# --------------------------------------------------------------------------- +# DataIntegrityValidator tests +# --------------------------------------------------------------------------- +class TestDataIntegrityValidator(unittest.TestCase): + def test_nan_detection(self): + self.assertTrue(DataIntegrityValidator.has_nan_or_inf(float("nan"))) + self.assertTrue(DataIntegrityValidator.has_nan_or_inf(float("inf"))) + self.assertTrue(DataIntegrityValidator.has_nan_or_inf(float("-inf"))) + self.assertFalse(DataIntegrityValidator.has_nan_or_inf(3.14)) + self.assertFalse(DataIntegrityValidator.has_nan_or_inf("string")) + + def test_ohlcv_clean_candle(self): + candle = {"open": 100, "high": 110, "low": 90, "close": 105, "volume": 1000} + violations = DataIntegrityValidator.check_ohlcv_consistency(candle) + self.assertEqual(violations, []) + + def test_ohlcv_high_less_than_low(self): + candle = {"open": 100, "high": 80, "low": 90, "close": 85} + violations = DataIntegrityValidator.check_ohlcv_consistency(candle) + self.assertTrue(any("high" in v and "low" in v for v in violations)) + + def test_ohlcv_close_above_high(self): + candle = {"open": 100, "high": 110, "low": 90, "close": 200} + violations = DataIntegrityValidator.check_ohlcv_consistency(candle) + self.assertTrue(any("close" in v for v in violations)) + + def test_ohlcv_negative_volume(self): + candle = {"open": 100, "high": 110, "low": 90, "close": 105, "volume": -50} + violations = DataIntegrityValidator.check_ohlcv_consistency(candle) + self.assertTrue(any("volume" in v for v in violations)) + + def test_ohlcv_nan_field(self): + candle = {"open": float("nan"), "high": 110, "low": 90, "close": 105} + violations = DataIntegrityValidator.check_ohlcv_consistency(candle) + self.assertTrue(any("NaN" in v for v in violations)) + + def test_detect_price_spikes(self): + prices = [100.0, 101.0, 99.0, 100.5, 10000.0, 100.2] # index 4 is spike + spikes = DataIntegrityValidator.detect_price_spikes(prices, z_threshold=2.0) + spike_indices = [i for i, _ in spikes] + self.assertIn(4, spike_indices) + + def test_detect_no_spikes_stable_series(self): + prices = [100.0 + i * 0.1 for i in range(20)] + spikes = DataIntegrityValidator.detect_price_spikes(prices, z_threshold=2.0) + self.assertEqual(spikes, []) + + def test_detect_price_spikes_too_short(self): + self.assertEqual(DataIntegrityValidator.detect_price_spikes([100.0, 200.0]), []) + + def test_validate_numeric_fields_clean(self): + data = {"price": 100.5, "volume": 1000.0} + violations = DataIntegrityValidator.validate_numeric_fields( + data, ["price", "volume"] + ) + self.assertEqual(violations, []) + + def test_validate_numeric_fields_missing(self): + data = {"price": 100.5} + violations = DataIntegrityValidator.validate_numeric_fields( + data, ["price", "volume"] + ) + self.assertTrue(any("volume" in v for v in violations)) + + def test_validate_numeric_fields_nan(self): + data = {"price": float("nan")} + violations = DataIntegrityValidator.validate_numeric_fields(data, ["price"]) + self.assertTrue(any("NaN" in v for v in violations)) + + def test_checksum_stable(self): + data = {"price": 100, "symbol": "BTC-USD"} + c1 = DataIntegrityValidator.compute_checksum(data) + c2 = DataIntegrityValidator.compute_checksum(data) + self.assertEqual(c1, c2) + + def test_checksum_verify(self): + data = {"price": 100} + cs = DataIntegrityValidator.compute_checksum(data) + self.assertTrue(DataIntegrityValidator.verify_checksum(data, cs)) + self.assertFalse(DataIntegrityValidator.verify_checksum({"price": 200}, cs)) + + def test_batch_integrity_all_clean(self): + records = [{"price": 100.0, "volume": 10.0} for _ in range(5)] + result = DataIntegrityValidator.check_batch_integrity( + records, ["price", "volume"] + ) + self.assertTrue(result["integrity_ok"]) + self.assertEqual(result["corrupt"], 0) + + def test_batch_integrity_detects_nan(self): + records = [ + {"price": 100.0, "volume": 10.0}, + {"price": float("nan"), "volume": 10.0}, + ] + result = DataIntegrityValidator.check_batch_integrity( + records, ["price", "volume"] + ) + self.assertFalse(result["integrity_ok"]) + self.assertIn(1, result["corrupt_indices"]) + + def test_batch_integrity_detects_missing_field(self): + records = [{"price": 100.0}, {"price": 200.0}] + result = DataIntegrityValidator.check_batch_integrity( + records, ["price", "volume"] + ) + self.assertFalse(result["integrity_ok"]) + + +if __name__ == "__main__": + unittest.main()