paulpas Role: Ensure trading data meets quality standards before processing to prevent bad decisions
Philosophy: Garbage in, garbage out; validating data at pipeline boundaries catches errors early and prevents cascading failures
Key Principles
- Schema Validation: Strict schema enforcement with typed dataclasses
- Range Checks: Validate values against expected ranges and business rules
- Consistency Checks: Ensure related data is coherent
- Missing Data Detection: Identify and handle missing or null values
- Data Quality Metrics: Track validation success rates and error patterns
Implementation Guidelines
Structure
- Core logic: validation/data_validator.py
- Schemas: validation/schemas.py
- Tests: tests/test_data_validation.py
Patterns to Follow
- Use Pydantic or dataclasses for schema definition
- Implement validation as pure functions
- Return detailed validation errors with context
- Support both synchronous and asynchronous validation
Adherence Checklist
Before completing your task, verify:
- All required fields are validated
- Range constraints are enforced
- Cross-field consistency is checked
- Validation errors include field paths
- Metrics are collected for monitoring
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
Python Implementation
import time
from typing import Dict, List, Optional, Any, Tuple
from dataclasses import dataclass, field
from enum import Enum
import logging
class ValidationErrorType(Enum):
MISSING = "missing"
INVALID_TYPE = "invalid_type"
OUT_OF_RANGE = "out_of_range"
CONSISTENCY = "consistency"
FORMAT = "format"
BUSINESS_RULE = "business_rule"
@dataclass
class ValidationError:
"""Represents a single validation error."""
field: str
error_type: ValidationErrorType
message: str
value: Any
context: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict:
return {
"field": self.field,
"error_type": self.error_type.value,
"message": self.message,
"value": self.value,
"context": self.context
}
@dataclass
class ValidationResult:
"""Result of data validation."""
is_valid: bool
errors: List[ValidationError] = field(default_factory=list)
warnings: List[ValidationError] = field(default_factory=list)
processed_at: float = field(default_factory=time.time)
def add_error(self, error: ValidationError):
"""Add error to validation result."""
self.errors.append(error)
self.is_valid = False
def add_warning(self, warning: ValidationError):
"""Add warning to validation result."""
self.warnings.append(warning)
def to_dict(self) -> Dict:
return {
"is_valid": self.is_valid,
"errors": [e.to_dict() for e in self.errors],
"warnings": [w.to_dict() for w in self.warnings],
"processed_at": self.processed_at
}
class DataValidator:
"""Validates trading data according to defined rules."""
def __init__(self):
self.validators: Dict[str, List[callable]] = {}
self._initialize_default_validators()
def _initialize_default_validators(self):
"""Initialize default validators for common data types."""
self.register_validator("price", self._validate_price)
self.register_validator("quantity", self._validate_quantity)
self.register_validator("timestamp", self._validate_timestamp)
self.register_validator("symbol", self._validate_symbol)
self.register_validator("candle", self._validate_candle)
self.register_validator("orderbook", self._validate_orderbook)
def register_validator(self, data_type: str, validator_func: callable):
"""Register a custom validator function."""
if data_type not in self.validators:
self.validators[data_type] = []
self.validators[data_type].append(validator_func)
def validate(self, data: Any, schema: str = None) -> ValidationResult:
"""Validate data against schema."""
result = ValidationResult(is_valid=True)
if schema and schema in self.validators:
for validator in self.validators[schema]:
try:
validation_result = validator(data, result)
if validation_result:
result.add_error(validation_result)
except Exception as e:
result.add_error(ValidationError(
field="validation",
error_type=ValidationErrorType.BUSINESS_RULE,
message=f"Validator error: {str(e)}",
value=str(validator)
))
return result
def _validate_price(self, data: Any, result: ValidationResult) -> Optional[ValidationError]:
"""Validate price value."""
if data is None:
return ValidationError(
field="price",
error_type=ValidationErrorType.MISSING,
message="Price is required",
value=None
)
if not isinstance(data, (int, float)):
return ValidationError(
field="price",
error_type=ValidationErrorType.INVALID_TYPE,
message="Price must be numeric",
value=data
)
if data <= 0:
return ValidationError(
field="price",
error_type=ValidationErrorType.OUT_OF_RANGE,
message="Price must be positive",
value=data
)
return None
def _validate_quantity(self, data: Any, result: ValidationResult) -> Optional[ValidationError]:
"""Validate quantity value."""
if data is None:
return ValidationError(
field="quantity",
error_type=ValidationErrorType.MISSING,
message="Quantity is required",
value=None
)
if not isinstance(data, (int, float)):
return ValidationError(
field="quantity",
error_type=ValidationErrorType.INVALID_TYPE,
message="Quantity must be numeric",
value=data
)
if data <= 0:
return ValidationError(
field="quantity",
error_type=ValidationErrorType.OUT_OF_RANGE,
message="Quantity must be positive",
value=data
)
return None
def _validate_timestamp(self, data: Any, result: ValidationResult) -> Optional[ValidationError]:
"""Validate timestamp value."""
if data is None:
return ValidationError(
field="timestamp",
error_type=ValidationErrorType.MISSING,
message="Timestamp is required",
value=None
)
if not isinstance(data, (int, float)):
return ValidationError(
field="timestamp",
error_type=ValidationErrorType.INVALID_TYPE,
message="Timestamp must be numeric",
value=data
)
# Check if timestamp is reasonable (within last year or next day)
current_time = time.time()
valid_range = 365 * 24 * 3600 # 1 year
if abs(data - current_time) > valid_range:
return ValidationError(
field="timestamp",
error_type=ValidationErrorType.OUT_OF_RANGE,
message="Timestamp appears invalid",
value=data,
context={"expected_range": f"±{valid_range}s from now"}
)
return None
def _validate_symbol(self, data: Any, result: ValidationResult) -> Optional[ValidationError]:
"""Validate symbol value."""
if data is None:
return ValidationError(
field="symbol",
error_type=ValidationErrorType.MISSING,
message="Symbol is required",
value=None
)
if not isinstance(data, str):
return ValidationError(
field="symbol",
error_type=ValidationErrorType.INVALID_TYPE,
message="Symbol must be string",
value=data
)
if not data or len(data) < 1:
return ValidationError(
field="symbol",
error_type=ValidationErrorType.FORMAT,
message="Symbol cannot be empty",
value=data
)
return None
def _validate_candle(self, data: Any, result: ValidationResult) -> Optional[ValidationError]:
"""Validate candle data structure."""
if not isinstance(data, dict):
return ValidationError(
field="candle",
error_type=ValidationErrorType.INVALID_TYPE,
message="Candle must be a dictionary",
value=data
)
required_fields = ["timestamp", "open", "high", "low", "close", "volume"]
for field in required_fields:
if field not in data:
return ValidationError(
field=f"candle.{field}",
error_type=ValidationErrorType.MISSING,
message=f"Required field missing: {field}",
value=data
)
# Validate individual fields
price_validation = self._validate_price(data.get("open"), result)
if price_validation:
return ValidationError(
field="candle.open",
error_type=ValidationErrorType.INVALID_TYPE,
message="Invalid open price",
value=data.get("open"),
context={"original_error": price_validation.message}
)
return None
def _validate_orderbook(self, data: Any, result: ValidationResult) -> Optional[ValidationError]:
"""Validate orderbook data structure."""
if not isinstance(data, dict):
return ValidationError(
field="orderbook",
error_type=ValidationErrorType.INVALID_TYPE,
message="Orderbook must be a dictionary",
value=data
)
if "bids" not in data or "asks" not in data:
return ValidationError(
field="orderbook",
error_type=ValidationErrorType.MISSING,
message="Orderbook must have bids and asks",
value=data
)
# Validate bids and asks are lists
for side in ["bids", "asks"]:
if not isinstance(data[side], list):
return ValidationError(
field=f"orderbook.{side}",
error_type=ValidationErrorType.INVALID_TYPE,
message=f"{side} must be a list",
value=data[side]
)
return None
class ValidationPipeline:
"""Pipeline for multi-stage data validation."""
def __init__(self, validators: List[DataValidator] = None):
self.validators = validators or []
self.metrics = {
"total_validated": 0,
"valid_count": 0,
"invalid_count": 0,
"warning_count": 0
}
def add_validator(self, validator: DataValidator):
"""Add validator to pipeline."""
self.validators.append(validator)
def validate(self, data: Any) -> ValidationResult:
"""Run validation through all validators."""
result = ValidationResult(is_valid=True)
self.metrics["total_validated"] += 1
for validator in self.validators:
validation = validator.validate(data)
if not validation.is_valid:
result.is_valid = False
for error in validation.errors:
result.add_error(error)
for warning in validation.warnings:
result.add_warning(warning)
self.metrics["warning_count"] += 1
if not validation.is_valid:
self.metrics["invalid_count"] += 1
else:
self.metrics["valid_count"] += 1
return result
def get_metrics(self) -> Dict:
"""Get validation metrics."""
return self.metrics.copy()
class BatchValidator:
"""Validates batches of data with aggregated reporting."""
def __init__(self, validator: DataValidator, batch_size: int = 1000):
self.validator = validator
self.batch_size = batch_size
self.results: List[ValidationResult] = []
def validate_batch(self, data_list: List[Any]) -> List[ValidationResult]:
"""Validate a batch of data items."""
results = []
for data in data_list:
result = self.validator.validate(data)
results.append(result)
self.results.append(result)
return results
def get_aggregate_report(self) -> Dict:
"""Get aggregate report of validation results."""
valid = sum(1 for r in self.results if r.is_valid)
invalid = len(self.results) - valid
total_errors = sum(len(r.errors) for r in self.results)
error_summary = {}
for result in self.results:
for error in result.errors:
key = f"{error.field}:{error.error_type.value}"
error_summary[key] = error_summary.get(key, 0) + 1
return {
"total_items": len(self.results),
"valid_items": valid,
"invalid_items": invalid,
"total_errors": total_errors,
"error_breakdown": error_summary,
"valid_rate": valid / len(self.results) if self.results else 0
}
Pattern 2: Schema Validation with Pydantic for Financial Data
from __future__ import annotations
import logging
from datetime import datetime, timezone
from decimal import Decimal
from typing import Optional
from pydantic import BaseModel, Field, field_validator
logger = logging.getLogger(__name__)
# --- Input Schemas (Parse Don't Validate — validated at boundary) ---
class TickData(BaseModel):
"""Validated tick record with strict constraints."""
symbol: str = Field(min_length=3, max_length=20, pattern=r'^[A-Z]+/[A-Z]+$')
price: Decimal = Field(gt=0)
size: Decimal = Field(gte=0)
side: str = Field(pattern=r'^(buy|sell)$')
timestamp_ms: int = Field(gt=0)
exchange: str = Field(min_length=1)
@field_validator("timestamp_ms")
@classmethod
def validate_timestamp_recent(cls, v: int) -> int:
now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
if now_ms - v > 30_000: # Stale by more than 30 seconds
raise ValueError("Timestamp is stale — data may be too old to trade on")
return v
class CandleData(BaseModel):
"""Validated OHLCV candle with cross-field validation."""
symbol: str
timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
open_price: Decimal = Field(gt=0)
high_price: Decimal
low_price: Decimal
close_price: Decimal = Field(gt=0)
volume: Decimal = Field(gte=0)
@field_validator("high_price")
@classmethod
def validate_high_not_below_prices(cls, v, info):
data = info.data
if v < data.get("open_price", 0) or v < data.get("low_price", 0):
raise ValueError(f"High ({v}) must be >= open and low")
return v
@field_validator("low_price")
@classmethod
def validate_low_not_above_prices(cls, v, info):
data = info.data
if v > data.get("open_price", 0) or v > data.get("high_price", 0):
raise ValueError(f"Low ({v}) must be <= open and high")
return v
class ValidationResult(BaseModel):
"""Result of validating a batch of records."""
total: int
valid: int
invalid: int
errors: list[dict] = Field(default_factory=list)
@property
def pass_rate(self) -> float:
return self.valid / self.total if self.total > 0 else 0.0
class DataValidator:
"""Validates batches of financial data against schema constraints."""
def validate_ticks(self, raw_records: list[dict]) -> ValidationResult:
results = ValidationResult(total=len(raw_records), valid=0, invalid=0)
for i, record in enumerate(raw_records):
try:
TickData(**record)
results.valid += 1
except Exception as e:
results.invalid += 1
results.errors.append({
"index": i,
"error": str(e),
"record_sample": {k: v for k, v in list(record.items())[:3]},
})
logger.info(
"Tick validation: %d/%d valid (%.1f%% pass rate)",
results.valid, results.total, results.pass_rate * 100,
)
return results
def validate_candles(self, raw_records: list[dict]) -> ValidationResult:
results = ValidationResult(total=len(raw_records), valid=0, invalid=0)
for i, record in enumerate(raw_records):
try:
CandleData(**record)
results.valid += 1
except Exception as e:
results.invalid += 1
results.errors.append({
"index": i,
"error": str(e),
})
logger.info(
"Candle validation: %d/%d valid (%.1f%% pass rate)",
results.valid, results.total, results.pass_rate * 100,
)
return results
Constraints
MUST DO
- Validate all incoming data against schema constraints (type, range, nullability) before processing or storage
- Implement idempotent operations: re-processing the same data must produce identical results
- Track data lineage and provenance with timestamps, source identifiers, and transformation history for every record
- Handle out-of-order data by implementing a watermark-based ordering mechanism with configurable tolerance window
- Log data quality metrics (completeness, freshness, accuracy) per source with automatic alerting on degradation
MUST NOT DO
- Do not silently drop records that fail validation — log them to a quarantine table for review
- Avoid concatenating strings for timestamp comparison; use proper datetime/timedelta objects
- Never assume data arrives in chronological order from any external feed without explicit ordering guarantees
- Do not store raw and processed data in the same table without clear partitioning or separation strategy
- Avoid blocking on slow data sources — implement async prefetch with timeout-based fallback to cached data
Live References
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.