From 822bfae7263e1f161605e90db5448495f3c3b928 Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Sun, 17 May 2026 23:35:53 +0300 Subject: [PATCH 1/5] feat: Add automated DB backup/restore and data corruption detection pt_backup.py: DatabaseBackupManager with SHA-256 verified SQLite backups, WAL checkpoint, PRAGMA integrity_check, point-in-time restore, retention pruning, and background scheduler. pt_validation.py: DataIntegrityValidator with NaN/Inf detection, OHLCV cross-field consistency checks, Z-score price spike detection, checksum verification, and batch integrity scanning. - 35 unit tests, all passing --- app/pt_backup.py | 399 ++++++++++++++++++++++++++++++++++ app/pt_validation.py | 380 ++++++++++++++++++-------------- app/test_backup_validation.py | 241 ++++++++++++++++++++ 3 files changed, 857 insertions(+), 163 deletions(-) create mode 100644 app/pt_backup.py create mode 100644 app/test_backup_validation.py diff --git a/app/pt_backup.py b/app/pt_backup.py new file mode 100644 index 00000000..f8144b33 --- /dev/null +++ b/app/pt_backup.py @@ -0,0 +1,399 @@ +""" +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 tempfile +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..53f34610 100644 --- a/app/pt_validation.py +++ b/app/pt_validation.py @@ -1,300 +1,201 @@ """ 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): """Custom exception for validation errors.""" + pass + +class DataCorruptionError(Exception): + """Raised when data integrity/corruption is detected.""" pass class InputValidator: """Comprehensive input validation for trading data.""" - # Valid cryptocurrency symbols pattern CRYPTO_SYMBOL_PATTERN = re.compile(r"^[A-Z]{2,10}$") - - # Valid trading pair pattern (e.g., BTC-USD, ETH-USDT) TRADING_PAIR_PATTERN = re.compile(r"^[A-Z]{2,10}-[A-Z]{2,10}$") - # Price validation limits (reasonable bounds for crypto prices) - MIN_PRICE = Decimal("0.00000001") # 1 satoshi equivalent - MAX_PRICE = Decimal("10000000") # 10 million USD - - # Volume validation limits + MIN_PRICE = Decimal("0.00000001") + MAX_PRICE = Decimal("10000000") MIN_VOLUME = Decimal("0.00000001") - MAX_VOLUME = Decimal("1000000000") # 1 billion units - - # Percentage limits + MAX_VOLUME = Decimal("1000000000") MIN_PERCENTAGE = Decimal("-100") - MAX_PERCENTAGE = Decimal("10000") # 100x gain max + MAX_PERCENTAGE = Decimal("10000") @staticmethod def validate_crypto_symbol(symbol: Any) -> str: - """Validate cryptocurrency symbol format.""" if not isinstance(symbol, str): raise ValidationError(f"Symbol must be string, got {type(symbol)}") - symbol = symbol.strip().upper() if not symbol: raise ValidationError("Symbol cannot be empty") - if not InputValidator.CRYPTO_SYMBOL_PATTERN.match(symbol): raise ValidationError(f"Invalid symbol format: {symbol}") - return symbol @staticmethod def validate_trading_pair(pair: Any) -> str: - """Validate trading pair format.""" if not isinstance(pair, str): raise ValidationError(f"Trading pair must be string, got {type(pair)}") - pair = pair.strip().upper() if not pair: raise ValidationError("Trading pair cannot be empty") - if not InputValidator.TRADING_PAIR_PATTERN.match(pair): raise ValidationError(f"Invalid trading pair format: {pair}") - return pair @staticmethod def validate_price(price: Any, field_name: str = "price") -> Decimal: - """Validate price value.""" try: if isinstance(price, str): price = price.strip() if not price: raise ValidationError(f"{field_name} cannot be empty") - decimal_price = Decimal(str(price)) - if decimal_price <= 0: raise ValidationError(f"{field_name} must be positive") - if decimal_price < InputValidator.MIN_PRICE: raise ValidationError(f"{field_name} too small: {decimal_price}") - if decimal_price > InputValidator.MAX_PRICE: raise ValidationError(f"{field_name} too large: {decimal_price}") - return decimal_price - - except (ValueError, InvalidOperation) as e: + except (ValueError, InvalidOperation): raise ValidationError(f"Invalid {field_name} format: {price}") @staticmethod def validate_volume(volume: Any, field_name: str = "volume") -> Decimal: - """Validate volume value.""" try: if isinstance(volume, str): volume = volume.strip() if not volume: raise ValidationError(f"{field_name} cannot be empty") - decimal_volume = Decimal(str(volume)) - if decimal_volume < 0: raise ValidationError(f"{field_name} cannot be negative") - if decimal_volume > InputValidator.MAX_VOLUME: raise ValidationError(f"{field_name} too large: {decimal_volume}") - return decimal_volume - - except (ValueError, InvalidOperation) as e: + except (ValueError, InvalidOperation): raise ValidationError(f"Invalid {field_name} format: {volume}") @staticmethod def validate_percentage(percentage: Any, field_name: str = "percentage") -> Decimal: - """Validate percentage value.""" try: if isinstance(percentage, str): percentage = percentage.strip().replace("%", "") if not percentage: raise ValidationError(f"{field_name} cannot be empty") - decimal_pct = Decimal(str(percentage)) - if decimal_pct < InputValidator.MIN_PERCENTAGE: raise ValidationError(f"{field_name} too low: {decimal_pct}%") - if decimal_pct > InputValidator.MAX_PERCENTAGE: raise ValidationError(f"{field_name} too high: {decimal_pct}%") - return decimal_pct - - except (ValueError, InvalidOperation) as e: + except (ValueError, InvalidOperation): raise ValidationError(f"Invalid {field_name} format: {percentage}") @staticmethod def validate_timestamp(timestamp: Any, field_name: str = "timestamp") -> int: - """Validate Unix timestamp.""" try: if isinstance(timestamp, str): timestamp = timestamp.strip() if not timestamp: raise ValidationError(f"{field_name} cannot be empty") - int_timestamp = int(float(timestamp)) - - # Reasonable bounds: 2020 to 2050 - if int_timestamp < 1577836800: # 2020-01-01 + if int_timestamp < 1577836800: raise ValidationError(f"{field_name} too old: {int_timestamp}") - - if int_timestamp > 2524608000: # 2050-01-01 - raise ValidationError( - f"{field_name} too far in future: {int_timestamp}" - ) - + if int_timestamp > 2524608000: + raise ValidationError(f"{field_name} too far in future: {int_timestamp}") return int_timestamp - - except (ValueError, TypeError) as e: + except (ValueError, TypeError): raise ValidationError(f"Invalid {field_name} format: {timestamp}") @staticmethod def validate_order_data(order_data: Dict[str, Any]) -> Dict[str, Any]: - """Validate trading order data structure.""" if not isinstance(order_data, dict): raise ValidationError("Order data must be a dictionary") - validated = {} - - # Required fields required_fields = ["id", "symbol", "price", "quantity", "side"] for field in required_fields: if field not in order_data: raise ValidationError(f"Missing required field: {field}") - - # Validate ID order_id = order_data.get("id") if not isinstance(order_id, (str, int)): raise ValidationError("Order ID must be string or integer") validated["id"] = str(order_id).strip() - - # Validate symbol validated["symbol"] = InputValidator.validate_trading_pair(order_data["symbol"]) - - # Validate price validated["price"] = InputValidator.validate_price(order_data["price"]) - - # Validate quantity - validated["quantity"] = InputValidator.validate_volume( - order_data["quantity"], "quantity" - ) - - # Validate side + validated["quantity"] = InputValidator.validate_volume(order_data["quantity"], "quantity") side = order_data.get("side", "").strip().lower() if side not in ["buy", "sell"]: raise ValidationError(f"Invalid order side: {side}") validated["side"] = side - - # Optional fields if "created_at" in order_data: - validated["created_at"] = InputValidator.validate_timestamp( - order_data["created_at"] - ) - + validated["created_at"] = InputValidator.validate_timestamp(order_data["created_at"]) if "status" in order_data: status = str(order_data["status"]).strip().lower() valid_statuses = ["pending", "filled", "cancelled", "rejected", "partial"] if status not in valid_statuses: raise ValidationError(f"Invalid order status: {status}") validated["status"] = status - return validated @staticmethod def validate_market_data(market_data: Dict[str, Any]) -> Dict[str, Any]: - """Validate market data structure from external APIs.""" if not isinstance(market_data, dict): raise ValidationError("Market data must be a dictionary") - validated = {} - - # Validate symbol if present if "symbol" in market_data: - validated["symbol"] = InputValidator.validate_trading_pair( - market_data["symbol"] - ) - - # Validate prices - price_fields = ["price", "open", "high", "low", "close", "ask", "bid"] - for field in price_fields: + validated["symbol"] = InputValidator.validate_trading_pair(market_data["symbol"]) + for field in ["price", "open", "high", "low", "close", "ask", "bid"]: if field in market_data and market_data[field] is not None: - validated[field] = InputValidator.validate_price( - market_data[field], field - ) - - # Validate volumes - volume_fields = ["volume", "base_volume", "quote_volume"] - for field in volume_fields: + validated[field] = InputValidator.validate_price(market_data[field], field) + for field in ["volume", "base_volume", "quote_volume"]: if field in market_data and market_data[field] is not None: - validated[field] = InputValidator.validate_volume( - market_data[field], field - ) - - # Validate timestamp + validated[field] = InputValidator.validate_volume(market_data[field], field) if "timestamp" in market_data: - validated["timestamp"] = InputValidator.validate_timestamp( - market_data["timestamp"] - ) - + validated["timestamp"] = InputValidator.validate_timestamp(market_data["timestamp"]) return validated @staticmethod def sanitize_string(input_str: Any, max_length: int = 1000) -> str: - """Sanitize string input to prevent injection attacks.""" if not isinstance(input_str, str): input_str = str(input_str) - - # Remove null bytes and control characters sanitized = "".join( char for char in input_str if ord(char) >= 32 or char in "\n\r\t" ) - - # Limit length - if len(sanitized) > max_length: - sanitized = sanitized[:max_length] - - return sanitized.strip() + return sanitized[:max_length].strip() @staticmethod def validate_config_data(config_data: Dict[str, Any]) -> Dict[str, Any]: - """Validate configuration file data.""" if not isinstance(config_data, dict): raise ValidationError("Config data must be a dictionary") - validated = {} - - # Validate coins list if "coins" in config_data: coins = config_data["coins"] if not isinstance(coins, list): raise ValidationError("Coins must be a list") - validated_coins = [] for coin in coins: try: - validated_coin = InputValidator.validate_crypto_symbol(coin) - validated_coins.append(validated_coin) + validated_coins.append(InputValidator.validate_crypto_symbol(coin)) except ValidationError: - continue # Skip invalid coins - + continue if not validated_coins: raise ValidationError("No valid coins found in configuration") - validated["coins"] = validated_coins - - # Validate numeric settings numeric_fields = { "trade_start_level": (1, 10), "start_allocation_pct": (0.0001, 1.0), @@ -304,60 +205,222 @@ def validate_config_data(config_data: Dict[str, Any]) -> Dict[str, Any]: "pm_start_pct_with_dca": (0.1, 100.0), "trailing_gap_pct": (0.1, 10.0), } - for field, (min_val, max_val) in numeric_fields.items(): if field in config_data: try: value = float(config_data[field]) if value < min_val or value > max_val: - raise ValidationError( - f"{field} must be between {min_val} and {max_val}" - ) + raise ValidationError(f"{field} must be between {min_val} and {max_val}") validated[field] = value except (ValueError, TypeError): raise ValidationError(f"Invalid {field} format") - - # Validate DCA levels if "dca_levels" in config_data: dca_levels = config_data["dca_levels"] if not isinstance(dca_levels, list): raise ValidationError("DCA levels must be a list") - validated_levels = [] for level in dca_levels: try: level_val = float(level) - if level_val > 0: # DCA levels should be negative (losses) + if level_val > 0: level_val = -level_val - if level_val < -90: # Reasonable limit + if level_val < -90: continue validated_levels.append(level_val) except (ValueError, TypeError): continue + validated["dca_levels"] = sorted(validated_levels, reverse=True) + if "main_neural_dir" in config_data and config_data["main_neural_dir"] is not None: + validated["main_neural_dir"] = InputValidator.sanitize_string( + config_data["main_neural_dir"], 500 + ) + return validated - validated["dca_levels"] = sorted( - validated_levels, reverse=True - ) # Sort from highest to lowest - # Validate directory path - if "main_neural_dir" in config_data: - dir_path = config_data["main_neural_dir"] - if dir_path is not None: - validated["main_neural_dir"] = InputValidator.sanitize_string( - dir_path, 500 - ) +# --------------------------------------------------------------------------- +# DataIntegrityValidator — corruption detection (Issue #63) +# --------------------------------------------------------------------------- +class DataIntegrityValidator: + """ + Detects data corruption and anomalies in trading and market data. - return validated + 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, + } + + +# --------------------------------------------------------------------------- +# Module-level helpers (unchanged public API) +# --------------------------------------------------------------------------- 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): raise ValidationError("JSON input must be string") - if len(json_str) > max_size: raise ValidationError(f"JSON too large: {len(json_str)} bytes") - try: data = json.loads(json_str) if not isinstance(data, dict): @@ -370,25 +433,17 @@ def safe_json_loads(json_str: str, max_size: int = 1024 * 1024) -> Dict[str, Any def validate_api_response( response_data: Any, expected_fields: List[str] = None ) -> Dict[str, Any]: - """Validate API response structure and content.""" if not isinstance(response_data, dict): raise ValidationError("API response must be a dictionary") - - # Check for error indicators if "error" in response_data: error_msg = InputValidator.sanitize_string(str(response_data["error"]), 200) raise ValidationError(f"API error: {error_msg}") - validated = {} - - # Validate expected fields if provided if expected_fields: for field in expected_fields: if field not in response_data: raise ValidationError(f"Missing expected field: {field}") validated[field] = response_data[field] - - # Add other fields with basic validation for key, value in response_data.items(): if key not in validated: if isinstance(value, str): @@ -396,6 +451,5 @@ def validate_api_response( elif isinstance(value, (int, float)): validated[key] = value elif isinstance(value, (list, dict)): - validated[key] = value # Keep as-is for complex structures - + validated[key] = value return validated diff --git a/app/test_backup_validation.py b/app/test_backup_validation.py new file mode 100644 index 00000000..71088198 --- /dev/null +++ b/app/test_backup_validation.py @@ -0,0 +1,241 @@ +"""Tests for pt_backup (issue #62) and DataIntegrityValidator (issue #63).""" + +import os +import sqlite3 +import tempfile +import time +import unittest + +from pt_backup import BackupRecord, DatabaseBackupManager +from pt_validation import DataCorruptionError, DataIntegrityValidator, ValidationError + + +# --------------------------------------------------------------------------- +# 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): + r1 = self.mgr.create_backup() + time.sleep(0.01) + r2 = 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() From 7d38e06d6afccf085300bb76117c65afc69effe4 Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Mon, 18 May 2026 03:56:14 +0300 Subject: [PATCH 2/5] fix: Remove unused imports and variables (flake8 F401/F841) --- app/pt_backup.py | 1 - app/pt_validation.py | 2 +- app/test_backup_validation.py | 8 ++++---- 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/app/pt_backup.py b/app/pt_backup.py index f8144b33..f968ad05 100644 --- a/app/pt_backup.py +++ b/app/pt_backup.py @@ -10,7 +10,6 @@ import os import shutil import sqlite3 -import tempfile import threading import time from dataclasses import dataclass, asdict diff --git a/app/pt_validation.py b/app/pt_validation.py index 53f34610..b174c9eb 100644 --- a/app/pt_validation.py +++ b/app/pt_validation.py @@ -8,7 +8,7 @@ import math import re from decimal import Decimal, InvalidOperation -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any, Dict, List, Tuple, Union class ValidationError(Exception): diff --git a/app/test_backup_validation.py b/app/test_backup_validation.py index 71088198..d2ad033d 100644 --- a/app/test_backup_validation.py +++ b/app/test_backup_validation.py @@ -6,8 +6,8 @@ import time import unittest -from pt_backup import BackupRecord, DatabaseBackupManager -from pt_validation import DataCorruptionError, DataIntegrityValidator, ValidationError +from pt_backup import DatabaseBackupManager +from pt_validation import DataIntegrityValidator # --------------------------------------------------------------------------- @@ -69,9 +69,9 @@ def test_list_backups(self): self.assertEqual(len(records), 2) def test_list_backups_sorted_oldest_first(self): - r1 = self.mgr.create_backup() + self.mgr.create_backup() time.sleep(0.01) - r2 = self.mgr.create_backup() + self.mgr.create_backup() records = self.mgr.list_backups() self.assertLessEqual(records[0].created_at, records[1].created_at) From 59b12848e7619f60d21fe46dc2735e582339e10a Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Mon, 18 May 2026 04:01:28 +0300 Subject: [PATCH 3/5] fix: Black formatting --- app/pt_backup.py | 55 ++++++++++++++++++++++++---------- app/pt_validation.py | 56 ++++++++++++++++++++++++++--------- app/test_backup_validation.py | 27 ++++++++++++----- 3 files changed, 100 insertions(+), 38 deletions(-) diff --git a/app/pt_backup.py b/app/pt_backup.py index f968ad05..184d2a82 100644 --- a/app/pt_backup.py +++ b/app/pt_backup.py @@ -25,6 +25,7 @@ @dataclass class BackupRecord: """Metadata for a single backup file.""" + backup_id: str source_db: str backup_path: str @@ -51,6 +52,7 @@ def from_dict(cls, d: dict) -> "BackupRecord": @dataclass class RestoreResult: """Result of a restore operation.""" + success: bool backup_id: str restored_from: str @@ -202,7 +204,9 @@ def create_backup(self, notes: str = "") -> Optional[BackupRecord]: self._append_manifest(record) logger.info( "Backup created: %s (%d bytes, integrity=%s)", - backup_id, file_size, integrity_ok, + backup_id, + file_size, + integrity_ok, ) # Step 7: Prune @@ -227,12 +231,16 @@ def verify_backup(self, backup_id: str) -> bool: if current_checksum != record.sha256_checksum: logger.error( "Backup checksum MISMATCH for %s: expected %s, got %s", - backup_id, record.sha256_checksum, current_checksum, + 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) + logger.info( + "Backup %s verified: checksum OK, integrity=%s", backup_id, integrity + ) return integrity # ------------------------------------------------------------------ @@ -250,8 +258,10 @@ def restore( record = self._get_record(backup_id) if not record: return RestoreResult( - success=False, backup_id=backup_id, - restored_from="", restored_to="", + success=False, + backup_id=backup_id, + restored_from="", + restored_to="", message=f"Backup record not found: {backup_id}", timestamp=time.time(), ) @@ -259,8 +269,10 @@ def restore( # 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="", + success=False, + backup_id=backup_id, + restored_from=record.backup_path, + restored_to="", message="Backup verification failed — restore aborted", timestamp=time.time(), ) @@ -274,8 +286,10 @@ def restore( 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, + success=True, + backup_id=backup_id, + restored_from=record.backup_path, + restored_to=target, message="Restore successful", timestamp=time.time(), ) @@ -286,8 +300,10 @@ def restore( pass logger.error("Restore failed: %s", exc) return RestoreResult( - success=False, backup_id=backup_id, - restored_from=record.backup_path, restored_to=target, + success=False, + backup_id=backup_id, + restored_from=record.backup_path, + restored_to=target, message=f"Restore failed: {exc}", timestamp=time.time(), ) @@ -297,8 +313,12 @@ def restore_latest(self, target_path: Optional[str] = None) -> RestoreResult: records = self.list_backups() if not records: return RestoreResult( - success=False, backup_id="", restored_from="", restored_to="", - message="No backups available", timestamp=time.time(), + success=False, + backup_id="", + restored_from="", + restored_to="", + message="No backups available", + timestamp=time.time(), ) return self.restore(records[-1].backup_id, target_path) @@ -330,12 +350,14 @@ def _append_manifest(self, record: BackupRecord) -> None: 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) + 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)] + to_prune = records[: max(0, len(records) - self.max_backups)] for record in to_prune: try: os.remove(record.backup_path) @@ -363,7 +385,8 @@ def start_scheduler(self) -> None: self._scheduler_thread.start() logger.info( "DB backup scheduler started (interval: %gh, max_backups: %d)", - self.backup_interval / 3600, self.max_backups, + self.backup_interval / 3600, + self.max_backups, ) def stop_scheduler(self) -> None: diff --git a/app/pt_validation.py b/app/pt_validation.py index b174c9eb..39c829b3 100644 --- a/app/pt_validation.py +++ b/app/pt_validation.py @@ -13,11 +13,13 @@ class ValidationError(Exception): """Custom exception for validation errors.""" + pass class DataCorruptionError(Exception): """Raised when data integrity/corruption is detected.""" + pass @@ -117,7 +119,9 @@ def validate_timestamp(timestamp: Any, field_name: str = "timestamp") -> int: if int_timestamp < 1577836800: raise ValidationError(f"{field_name} too old: {int_timestamp}") if int_timestamp > 2524608000: - raise ValidationError(f"{field_name} too far in future: {int_timestamp}") + raise ValidationError( + f"{field_name} too far in future: {int_timestamp}" + ) return int_timestamp except (ValueError, TypeError): raise ValidationError(f"Invalid {field_name} format: {timestamp}") @@ -137,13 +141,17 @@ def validate_order_data(order_data: Dict[str, Any]) -> Dict[str, Any]: validated["id"] = str(order_id).strip() validated["symbol"] = InputValidator.validate_trading_pair(order_data["symbol"]) validated["price"] = InputValidator.validate_price(order_data["price"]) - validated["quantity"] = InputValidator.validate_volume(order_data["quantity"], "quantity") + validated["quantity"] = InputValidator.validate_volume( + order_data["quantity"], "quantity" + ) side = order_data.get("side", "").strip().lower() if side not in ["buy", "sell"]: raise ValidationError(f"Invalid order side: {side}") validated["side"] = side if "created_at" in order_data: - validated["created_at"] = InputValidator.validate_timestamp(order_data["created_at"]) + validated["created_at"] = InputValidator.validate_timestamp( + order_data["created_at"] + ) if "status" in order_data: status = str(order_data["status"]).strip().lower() valid_statuses = ["pending", "filled", "cancelled", "rejected", "partial"] @@ -158,15 +166,23 @@ def validate_market_data(market_data: Dict[str, Any]) -> Dict[str, Any]: raise ValidationError("Market data must be a dictionary") validated = {} if "symbol" in market_data: - validated["symbol"] = InputValidator.validate_trading_pair(market_data["symbol"]) + validated["symbol"] = InputValidator.validate_trading_pair( + market_data["symbol"] + ) for field in ["price", "open", "high", "low", "close", "ask", "bid"]: if field in market_data and market_data[field] is not None: - validated[field] = InputValidator.validate_price(market_data[field], field) + validated[field] = InputValidator.validate_price( + market_data[field], field + ) for field in ["volume", "base_volume", "quote_volume"]: if field in market_data and market_data[field] is not None: - validated[field] = InputValidator.validate_volume(market_data[field], field) + validated[field] = InputValidator.validate_volume( + market_data[field], field + ) if "timestamp" in market_data: - validated["timestamp"] = InputValidator.validate_timestamp(market_data["timestamp"]) + validated["timestamp"] = InputValidator.validate_timestamp( + market_data["timestamp"] + ) return validated @staticmethod @@ -210,7 +226,9 @@ def validate_config_data(config_data: Dict[str, Any]) -> Dict[str, Any]: try: value = float(config_data[field]) if value < min_val or value > max_val: - raise ValidationError(f"{field} must be between {min_val} and {max_val}") + raise ValidationError( + f"{field} must be between {min_val} and {max_val}" + ) validated[field] = value except (ValueError, TypeError): raise ValidationError(f"Invalid {field} format") @@ -230,7 +248,10 @@ def validate_config_data(config_data: Dict[str, Any]) -> Dict[str, Any]: except (ValueError, TypeError): continue validated["dca_levels"] = sorted(validated_levels, reverse=True) - if "main_neural_dir" in config_data and config_data["main_neural_dir"] is not None: + if ( + "main_neural_dir" in config_data + and config_data["main_neural_dir"] is not None + ): validated["main_neural_dir"] = InputValidator.sanitize_string( config_data["main_neural_dir"], 500 ) @@ -253,7 +274,7 @@ class DataIntegrityValidator: """ # Default anomaly thresholds - PRICE_SPIKE_Z_SCORE = 5.0 # Flag if value deviates > 5 std devs from series mean + 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 @@ -285,13 +306,20 @@ def check_ohlcv_consistency(candle: Dict[str, Any]) -> List[str]: 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") + 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 + 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})") @@ -323,6 +351,7 @@ def detect_price_spikes( if len(prices) < 3: return [] import statistics + try: mean = statistics.mean(prices) std = statistics.stdev(prices) @@ -331,8 +360,7 @@ def detect_price_spikes( if std == 0: return [] return [ - (i, v) for i, v in enumerate(prices) - if abs((v - mean) / std) > z_threshold + (i, v) for i, v in enumerate(prices) if abs((v - mean) / std) > z_threshold ] @staticmethod diff --git a/app/test_backup_validation.py b/app/test_backup_validation.py index d2ad033d..f4739b8b 100644 --- a/app/test_backup_validation.py +++ b/app/test_backup_validation.py @@ -16,14 +16,15 @@ 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( + "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") @@ -135,6 +136,7 @@ def test_backup_manifest_persists(self): def tearDown(self): import shutil + shutil.rmtree(self.tmpdir, ignore_errors=True) @@ -142,7 +144,6 @@ def tearDown(self): # 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"))) @@ -191,12 +192,16 @@ def test_detect_price_spikes_too_short(self): def test_validate_numeric_fields_clean(self): data = {"price": 100.5, "volume": 1000.0} - violations = DataIntegrityValidator.validate_numeric_fields(data, ["price", "volume"]) + 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"]) + 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): @@ -218,7 +223,9 @@ def test_checksum_verify(self): 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"]) + result = DataIntegrityValidator.check_batch_integrity( + records, ["price", "volume"] + ) self.assertTrue(result["integrity_ok"]) self.assertEqual(result["corrupt"], 0) @@ -227,13 +234,17 @@ def test_batch_integrity_detects_nan(self): {"price": 100.0, "volume": 10.0}, {"price": float("nan"), "volume": 10.0}, ] - result = DataIntegrityValidator.check_batch_integrity(records, ["price", "volume"]) + 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"]) + result = DataIntegrityValidator.check_batch_integrity( + records, ["price", "volume"] + ) self.assertFalse(result["integrity_ok"]) From cb116066223f49677bd814186dbeb2ffcc5cde51 Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Tue, 19 May 2026 11:47:41 +0300 Subject: [PATCH 4/5] style: apply Black formatting to pt_validation.py Black check failed in CI due to rebase conflict resolution leaving pt_validation.py unformatted. Auto-formatted with Black. --- app/pt_validation.py | 1 + 1 file changed, 1 insertion(+) diff --git a/app/pt_validation.py b/app/pt_validation.py index 39c829b3..358f0d68 100644 --- a/app/pt_validation.py +++ b/app/pt_validation.py @@ -3,6 +3,7 @@ Provides comprehensive validation for external data sources, user inputs, and data corruption/anomaly detection. """ + import hashlib import json import math From 05c2a164bd5e88cce89f1569e58f27e783053c14 Mon Sep 17 00:00:00 2001 From: Ibrahim Elsayed Date: Sat, 23 May 2026 16:24:49 +0300 Subject: [PATCH 5/5] fix: restore stripped docstrings/comments in InputValidator PR82 review feedback: the prior commit removed docstrings and inline comments from existing InputValidator methods that were unrelated to the data-integrity additions. Rebuild pt_validation.py from main and re-apply only the issue-scoped additions (DataCorruptionError, DataIntegrityValidator, hashlib/math/Tuple imports, module docstring update). --- app/pt_validation.py | 171 +++++++++++++++++++++++++++++++++++-------- 1 file changed, 140 insertions(+), 31 deletions(-) diff --git a/app/pt_validation.py b/app/pt_validation.py index 358f0d68..1874b99a 100644 --- a/app/pt_validation.py +++ b/app/pt_validation.py @@ -9,7 +9,7 @@ import math import re from decimal import Decimal, InvalidOperation -from typing import Any, Dict, List, Tuple, Union +from typing import Any, Dict, List, Optional, Tuple, Union class ValidationError(Exception): @@ -27,192 +27,283 @@ class DataCorruptionError(Exception): class InputValidator: """Comprehensive input validation for trading data.""" + # Valid cryptocurrency symbols pattern CRYPTO_SYMBOL_PATTERN = re.compile(r"^[A-Z]{2,10}$") + + # Valid trading pair pattern (e.g., BTC-USD, ETH-USDT) TRADING_PAIR_PATTERN = re.compile(r"^[A-Z]{2,10}-[A-Z]{2,10}$") - MIN_PRICE = Decimal("0.00000001") - MAX_PRICE = Decimal("10000000") + # Price validation limits (reasonable bounds for crypto prices) + MIN_PRICE = Decimal("0.00000001") # 1 satoshi equivalent + MAX_PRICE = Decimal("10000000") # 10 million USD + + # Volume validation limits MIN_VOLUME = Decimal("0.00000001") - MAX_VOLUME = Decimal("1000000000") + MAX_VOLUME = Decimal("1000000000") # 1 billion units + + # Percentage limits MIN_PERCENTAGE = Decimal("-100") - MAX_PERCENTAGE = Decimal("10000") + MAX_PERCENTAGE = Decimal("10000") # 100x gain max @staticmethod def validate_crypto_symbol(symbol: Any) -> str: + """Validate cryptocurrency symbol format.""" if not isinstance(symbol, str): raise ValidationError(f"Symbol must be string, got {type(symbol)}") + symbol = symbol.strip().upper() if not symbol: raise ValidationError("Symbol cannot be empty") + if not InputValidator.CRYPTO_SYMBOL_PATTERN.match(symbol): raise ValidationError(f"Invalid symbol format: {symbol}") + return symbol @staticmethod def validate_trading_pair(pair: Any) -> str: + """Validate trading pair format.""" if not isinstance(pair, str): raise ValidationError(f"Trading pair must be string, got {type(pair)}") + pair = pair.strip().upper() if not pair: raise ValidationError("Trading pair cannot be empty") + if not InputValidator.TRADING_PAIR_PATTERN.match(pair): raise ValidationError(f"Invalid trading pair format: {pair}") + return pair @staticmethod def validate_price(price: Any, field_name: str = "price") -> Decimal: + """Validate price value.""" try: if isinstance(price, str): price = price.strip() if not price: raise ValidationError(f"{field_name} cannot be empty") + decimal_price = Decimal(str(price)) + if decimal_price <= 0: raise ValidationError(f"{field_name} must be positive") + if decimal_price < InputValidator.MIN_PRICE: raise ValidationError(f"{field_name} too small: {decimal_price}") + if decimal_price > InputValidator.MAX_PRICE: raise ValidationError(f"{field_name} too large: {decimal_price}") + return decimal_price - except (ValueError, InvalidOperation): + + except (ValueError, InvalidOperation) as e: raise ValidationError(f"Invalid {field_name} format: {price}") @staticmethod def validate_volume(volume: Any, field_name: str = "volume") -> Decimal: + """Validate volume value.""" try: if isinstance(volume, str): volume = volume.strip() if not volume: raise ValidationError(f"{field_name} cannot be empty") + decimal_volume = Decimal(str(volume)) + if decimal_volume < 0: raise ValidationError(f"{field_name} cannot be negative") + if decimal_volume > InputValidator.MAX_VOLUME: raise ValidationError(f"{field_name} too large: {decimal_volume}") + return decimal_volume - except (ValueError, InvalidOperation): + + except (ValueError, InvalidOperation) as e: raise ValidationError(f"Invalid {field_name} format: {volume}") @staticmethod def validate_percentage(percentage: Any, field_name: str = "percentage") -> Decimal: + """Validate percentage value.""" try: if isinstance(percentage, str): percentage = percentage.strip().replace("%", "") if not percentage: raise ValidationError(f"{field_name} cannot be empty") + decimal_pct = Decimal(str(percentage)) + if decimal_pct < InputValidator.MIN_PERCENTAGE: raise ValidationError(f"{field_name} too low: {decimal_pct}%") + if decimal_pct > InputValidator.MAX_PERCENTAGE: raise ValidationError(f"{field_name} too high: {decimal_pct}%") + return decimal_pct - except (ValueError, InvalidOperation): + + except (ValueError, InvalidOperation) as e: raise ValidationError(f"Invalid {field_name} format: {percentage}") @staticmethod def validate_timestamp(timestamp: Any, field_name: str = "timestamp") -> int: + """Validate Unix timestamp.""" try: if isinstance(timestamp, str): timestamp = timestamp.strip() if not timestamp: raise ValidationError(f"{field_name} cannot be empty") + int_timestamp = int(float(timestamp)) - if int_timestamp < 1577836800: + + # Reasonable bounds: 2020 to 2050 + if int_timestamp < 1577836800: # 2020-01-01 raise ValidationError(f"{field_name} too old: {int_timestamp}") - if int_timestamp > 2524608000: + + if int_timestamp > 2524608000: # 2050-01-01 raise ValidationError( f"{field_name} too far in future: {int_timestamp}" ) + return int_timestamp - except (ValueError, TypeError): + + except (ValueError, TypeError) as e: raise ValidationError(f"Invalid {field_name} format: {timestamp}") @staticmethod def validate_order_data(order_data: Dict[str, Any]) -> Dict[str, Any]: + """Validate trading order data structure.""" if not isinstance(order_data, dict): raise ValidationError("Order data must be a dictionary") + validated = {} + + # Required fields required_fields = ["id", "symbol", "price", "quantity", "side"] for field in required_fields: if field not in order_data: raise ValidationError(f"Missing required field: {field}") + + # Validate ID order_id = order_data.get("id") if not isinstance(order_id, (str, int)): raise ValidationError("Order ID must be string or integer") validated["id"] = str(order_id).strip() + + # Validate symbol validated["symbol"] = InputValidator.validate_trading_pair(order_data["symbol"]) + + # Validate price validated["price"] = InputValidator.validate_price(order_data["price"]) + + # Validate quantity validated["quantity"] = InputValidator.validate_volume( order_data["quantity"], "quantity" ) + + # Validate side side = order_data.get("side", "").strip().lower() if side not in ["buy", "sell"]: raise ValidationError(f"Invalid order side: {side}") validated["side"] = side + + # Optional fields if "created_at" in order_data: validated["created_at"] = InputValidator.validate_timestamp( order_data["created_at"] ) + if "status" in order_data: status = str(order_data["status"]).strip().lower() valid_statuses = ["pending", "filled", "cancelled", "rejected", "partial"] if status not in valid_statuses: raise ValidationError(f"Invalid order status: {status}") validated["status"] = status + return validated @staticmethod def validate_market_data(market_data: Dict[str, Any]) -> Dict[str, Any]: + """Validate market data structure from external APIs.""" if not isinstance(market_data, dict): raise ValidationError("Market data must be a dictionary") + validated = {} + + # Validate symbol if present if "symbol" in market_data: validated["symbol"] = InputValidator.validate_trading_pair( market_data["symbol"] ) - for field in ["price", "open", "high", "low", "close", "ask", "bid"]: + + # Validate prices + price_fields = ["price", "open", "high", "low", "close", "ask", "bid"] + for field in price_fields: if field in market_data and market_data[field] is not None: validated[field] = InputValidator.validate_price( market_data[field], field ) - for field in ["volume", "base_volume", "quote_volume"]: + + # Validate volumes + volume_fields = ["volume", "base_volume", "quote_volume"] + for field in volume_fields: if field in market_data and market_data[field] is not None: validated[field] = InputValidator.validate_volume( market_data[field], field ) + + # Validate timestamp if "timestamp" in market_data: validated["timestamp"] = InputValidator.validate_timestamp( market_data["timestamp"] ) + return validated @staticmethod def sanitize_string(input_str: Any, max_length: int = 1000) -> str: + """Sanitize string input to prevent injection attacks.""" if not isinstance(input_str, str): input_str = str(input_str) + + # Remove null bytes and control characters sanitized = "".join( char for char in input_str if ord(char) >= 32 or char in "\n\r\t" ) - return sanitized[:max_length].strip() + + # Limit length + if len(sanitized) > max_length: + sanitized = sanitized[:max_length] + + return sanitized.strip() @staticmethod def validate_config_data(config_data: Dict[str, Any]) -> Dict[str, Any]: + """Validate configuration file data.""" if not isinstance(config_data, dict): raise ValidationError("Config data must be a dictionary") + validated = {} + + # Validate coins list if "coins" in config_data: coins = config_data["coins"] if not isinstance(coins, list): raise ValidationError("Coins must be a list") + validated_coins = [] for coin in coins: try: - validated_coins.append(InputValidator.validate_crypto_symbol(coin)) + validated_coin = InputValidator.validate_crypto_symbol(coin) + validated_coins.append(validated_coin) except ValidationError: - continue + continue # Skip invalid coins + if not validated_coins: raise ValidationError("No valid coins found in configuration") + validated["coins"] = validated_coins + + # Validate numeric settings numeric_fields = { "trade_start_level": (1, 10), "start_allocation_pct": (0.0001, 1.0), @@ -222,6 +313,7 @@ def validate_config_data(config_data: Dict[str, Any]) -> Dict[str, Any]: "pm_start_pct_with_dca": (0.1, 100.0), "trailing_gap_pct": (0.1, 10.0), } + for field, (min_val, max_val) in numeric_fields.items(): if field in config_data: try: @@ -233,29 +325,37 @@ def validate_config_data(config_data: Dict[str, Any]) -> Dict[str, Any]: validated[field] = value except (ValueError, TypeError): raise ValidationError(f"Invalid {field} format") + + # Validate DCA levels if "dca_levels" in config_data: dca_levels = config_data["dca_levels"] if not isinstance(dca_levels, list): raise ValidationError("DCA levels must be a list") + validated_levels = [] for level in dca_levels: try: level_val = float(level) - if level_val > 0: + if level_val > 0: # DCA levels should be negative (losses) level_val = -level_val - if level_val < -90: + if level_val < -90: # Reasonable limit continue validated_levels.append(level_val) except (ValueError, TypeError): continue - validated["dca_levels"] = sorted(validated_levels, reverse=True) - if ( - "main_neural_dir" in config_data - and config_data["main_neural_dir"] is not None - ): - validated["main_neural_dir"] = InputValidator.sanitize_string( - config_data["main_neural_dir"], 500 - ) + + validated["dca_levels"] = sorted( + validated_levels, reverse=True + ) # Sort from highest to lowest + + # Validate directory path + if "main_neural_dir" in config_data: + dir_path = config_data["main_neural_dir"] + if dir_path is not None: + validated["main_neural_dir"] = InputValidator.sanitize_string( + dir_path, 500 + ) + return validated @@ -280,7 +380,7 @@ class DataIntegrityValidator: @staticmethod def has_nan_or_inf(value: Any) -> bool: - """True if value is float NaN or ±Infinity.""" + """True if value is float NaN or +/-Infinity.""" if isinstance(value, float): return math.isnan(value) or math.isinf(value) if isinstance(value, Decimal): @@ -442,14 +542,14 @@ def check_batch_integrity( } -# --------------------------------------------------------------------------- -# Module-level helpers (unchanged public API) -# --------------------------------------------------------------------------- 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): raise ValidationError("JSON input must be string") + if len(json_str) > max_size: raise ValidationError(f"JSON too large: {len(json_str)} bytes") + try: data = json.loads(json_str) if not isinstance(data, dict): @@ -462,17 +562,25 @@ def safe_json_loads(json_str: str, max_size: int = 1024 * 1024) -> Dict[str, Any def validate_api_response( response_data: Any, expected_fields: List[str] = None ) -> Dict[str, Any]: + """Validate API response structure and content.""" if not isinstance(response_data, dict): raise ValidationError("API response must be a dictionary") + + # Check for error indicators if "error" in response_data: error_msg = InputValidator.sanitize_string(str(response_data["error"]), 200) raise ValidationError(f"API error: {error_msg}") + validated = {} + + # Validate expected fields if provided if expected_fields: for field in expected_fields: if field not in response_data: raise ValidationError(f"Missing expected field: {field}") validated[field] = response_data[field] + + # Add other fields with basic validation for key, value in response_data.items(): if key not in validated: if isinstance(value, str): @@ -480,5 +588,6 @@ def validate_api_response( elif isinstance(value, (int, float)): validated[key] = value elif isinstance(value, (list, dict)): - validated[key] = value + validated[key] = value # Keep as-is for complex structures + return validated