Spaces:
Runtime error
Runtime error
Matthew-Anyiam
Add missing source files: 13F holdings, validator, transcripts, realtime, bulk
f8e5e3b | """ | |
| Eugene Intelligence - XBRL Data Validator | |
| Comprehensive validation layer for financial metrics to prevent bad SEC data reaching agents. | |
| """ | |
| import json | |
| import logging | |
| import statistics | |
| from typing import Dict, List, Any, Optional, Tuple | |
| from dataclasses import dataclass | |
| from datetime import datetime, timedelta | |
| from eugene.config import Config, get_config | |
| logger = logging.getLogger(__name__) | |
| class ValidationResult: | |
| """Result of a single validation check.""" | |
| check_name: str | |
| status: str # 'pass', 'fail', 'warning' | |
| message: str | |
| severity: str # 'low', 'medium', 'high', 'critical' | |
| class ValidatedMetric: | |
| """A financial metric with validation results and confidence scoring.""" | |
| value: Any | |
| raw_value: Any | |
| xbrl_tag: str | |
| source: str | |
| confidence: int # 0-100 | |
| validation: Dict[str, str] # check_name -> status | |
| flags: List[str] | |
| validation_results: List[ValidationResult] | |
| def to_dict(self): | |
| return { | |
| "value": self.value, | |
| "raw_value": self.raw_value, | |
| "xbrl_tag": self.xbrl_tag, | |
| "source": self.source, | |
| "confidence": self.confidence, | |
| "validation": self.validation, | |
| "flags": self.flags | |
| } | |
| class XBRLValidator: | |
| """Validates XBRL financial metrics before they reach agents.""" | |
| def __init__(self, config=None): | |
| self.config = config or get_config() | |
| # Critical metrics that should never be negative | |
| self.positive_only_metrics = { | |
| 'revenue', 'total_assets', 'cash_and_equivalents', 'shares_outstanding' | |
| } | |
| # Metrics where negative values might be acceptable but worth flagging | |
| self.usually_positive_metrics = { | |
| 'operating_income', 'net_income', 'operating_cash_flow', 'total_equity' | |
| } | |
| # Source quality scoring | |
| self.source_weights = { | |
| '10-K': 100, | |
| '10-Q': 85, | |
| '8-K': 70, | |
| '10-K/A': 95, # amended | |
| '10-Q/A': 80 # amended | |
| } | |
| def validate_financials(self, xbrl_financials, ticker=None) -> Dict[str, ValidatedMetric]: | |
| """ | |
| Validate all financial metrics in an XBRLFinancials object. | |
| Args: | |
| xbrl_financials: XBRLFinancials object from XBRL client | |
| ticker: Optional ticker for historical comparison | |
| Returns: | |
| Dict of metric_name -> ValidatedMetric | |
| """ | |
| validated_metrics = {} | |
| # Get historical data for trend analysis if ticker provided | |
| historical_data = self._get_historical_context(ticker) if ticker else {} | |
| for metric_name, fact in xbrl_financials.facts.items(): | |
| validated_metric = self._validate_single_metric( | |
| metric_name, fact, historical_data.get(metric_name, []) | |
| ) | |
| validated_metrics[metric_name] = validated_metric | |
| return validated_metrics | |
| def _validate_single_metric(self, metric_name: str, fact, historical_values: List[float]) -> ValidatedMetric: | |
| """Validate a single financial metric.""" | |
| validation_results = [] | |
| flags = [] | |
| # 1. Magnitude Check - flag >100x YoY changes | |
| magnitude_result = self._check_magnitude(metric_name, fact.value, historical_values) | |
| validation_results.append(magnitude_result) | |
| if magnitude_result.status != 'pass': | |
| flags.append(f"Magnitude {magnitude_result.status}: {magnitude_result.message}") | |
| # 2. Sign Check - flag negative values for metrics that should be positive | |
| sign_result = self._check_sign(metric_name, fact.value) | |
| validation_results.append(sign_result) | |
| if sign_result.status != 'pass': | |
| flags.append(f"Sign {sign_result.status}: {sign_result.message}") | |
| # 3. Historical Outlier Check - flag >3 standard deviations from history | |
| outlier_result = self._check_historical_outlier(metric_name, fact.value, historical_values) | |
| validation_results.append(outlier_result) | |
| if outlier_result.status != 'pass': | |
| flags.append(f"Outlier {outlier_result.status}: {outlier_result.message}") | |
| # 4. Data Completeness Check | |
| completeness_result = self._check_data_completeness(fact) | |
| validation_results.append(completeness_result) | |
| if completeness_result.status != 'pass': | |
| flags.append(f"Completeness {completeness_result.status}: {completeness_result.message}") | |
| # Calculate confidence score | |
| confidence = self._calculate_confidence(fact, validation_results) | |
| # Build validation summary | |
| validation_summary = { | |
| result.check_name: result.status for result in validation_results | |
| } | |
| # Format source string | |
| source_str = f"{fact.form} {fact.filed}" if fact.form and fact.filed else "Unknown source" | |
| return ValidatedMetric( | |
| value=fact.value, | |
| raw_value=fact.value, | |
| xbrl_tag=fact.tag, | |
| source=source_str, | |
| confidence=confidence, | |
| validation=validation_summary, | |
| flags=flags, | |
| validation_results=validation_results | |
| ) | |
| def _check_magnitude(self, metric_name: str, current_value: float, historical_values: List[float]) -> ValidationResult: | |
| """Check for extreme year-over-year changes (>100x).""" | |
| if not historical_values or current_value is None: | |
| return ValidationResult( | |
| check_name="magnitude_check", | |
| status="pass", | |
| message="No historical data for comparison", | |
| severity="low" | |
| ) | |
| # Compare to most recent historical value | |
| recent_value = historical_values[0] if historical_values else None | |
| if recent_value is None or recent_value == 0: | |
| return ValidationResult( | |
| check_name="magnitude_check", | |
| status="warning", | |
| message="Cannot compare to zero/null historical value", | |
| severity="low" | |
| ) | |
| ratio = abs(current_value / recent_value) | |
| reverse_ratio = abs(recent_value / current_value) | |
| # Check both directions (increase and decrease) | |
| max_ratio = max(ratio, reverse_ratio) | |
| if max_ratio > 100: | |
| return ValidationResult( | |
| check_name="magnitude_check", | |
| status="fail", | |
| message=f"Extreme change: {max_ratio:.0f}x from previous period", | |
| severity="critical" | |
| ) | |
| elif max_ratio >= 10: | |
| return ValidationResult( | |
| check_name="magnitude_check", | |
| status="warning", | |
| message=f"Large change: {max_ratio:.1f}x from previous period", | |
| severity="medium" | |
| ) | |
| else: | |
| return ValidationResult( | |
| check_name="magnitude_check", | |
| status="pass", | |
| message="Normal magnitude change", | |
| severity="low" | |
| ) | |
| def _check_sign(self, metric_name: str, value: float) -> ValidationResult: | |
| """Check if metric has expected sign (positive/negative).""" | |
| if value is None: | |
| return ValidationResult( | |
| check_name="sign_check", | |
| status="warning", | |
| message="Null value", | |
| severity="medium" | |
| ) | |
| if metric_name in self.positive_only_metrics and value < 0: | |
| return ValidationResult( | |
| check_name="sign_check", | |
| status="fail", | |
| message=f"{metric_name} should not be negative: {value}", | |
| severity="high" | |
| ) | |
| if metric_name in self.usually_positive_metrics and value < 0: | |
| return ValidationResult( | |
| check_name="sign_check", | |
| status="warning", | |
| message=f"{metric_name} is negative, worth reviewing: {value}", | |
| severity="medium" | |
| ) | |
| return ValidationResult( | |
| check_name="sign_check", | |
| status="pass", | |
| message="Expected sign", | |
| severity="low" | |
| ) | |
| def _check_historical_outlier(self, metric_name: str, current_value: float, historical_values: List[float]) -> ValidationResult: | |
| """Check if current value is >3 standard deviations from historical mean.""" | |
| if len(historical_values) < 3 or current_value is None: | |
| return ValidationResult( | |
| check_name="outlier_check", | |
| status="pass", | |
| message="Insufficient historical data", | |
| severity="low" | |
| ) | |
| # Filter out None/null values | |
| clean_values = [v for v in historical_values if v is not None] | |
| if len(clean_values) < 3: | |
| return ValidationResult( | |
| check_name="outlier_check", | |
| status="pass", | |
| message="Insufficient clean historical data", | |
| severity="low" | |
| ) | |
| try: | |
| mean = statistics.mean(clean_values) | |
| if len(clean_values) == 1: | |
| return ValidationResult( | |
| check_name="outlier_check", | |
| status="pass", | |
| message="Only one historical value", | |
| severity="low" | |
| ) | |
| stdev = statistics.stdev(clean_values) | |
| if stdev == 0: | |
| # All historical values are identical | |
| if current_value != mean: | |
| return ValidationResult( | |
| check_name="outlier_check", | |
| status="warning", | |
| message=f"Value {current_value} differs from constant historical {mean}", | |
| severity="medium" | |
| ) | |
| else: | |
| return ValidationResult( | |
| check_name="outlier_check", | |
| status="pass", | |
| message="Consistent with historical values", | |
| severity="low" | |
| ) | |
| z_score = abs((current_value - mean) / stdev) | |
| if z_score > 3: | |
| return ValidationResult( | |
| check_name="outlier_check", | |
| status="fail", | |
| message=f"Extreme outlier: {z_score:.1f} std devs from mean", | |
| severity="high" | |
| ) | |
| elif z_score > 2: | |
| return ValidationResult( | |
| check_name="outlier_check", | |
| status="warning", | |
| message=f"Potential outlier: {z_score:.1f} std devs from mean", | |
| severity="medium" | |
| ) | |
| else: | |
| return ValidationResult( | |
| check_name="outlier_check", | |
| status="pass", | |
| message="Within normal range", | |
| severity="low" | |
| ) | |
| except statistics.StatisticsError as e: | |
| return ValidationResult( | |
| check_name="outlier_check", | |
| status="warning", | |
| message=f"Statistical analysis failed: {e}", | |
| severity="low" | |
| ) | |
| def _check_data_completeness(self, fact) -> ValidationResult: | |
| """Check if the XBRL fact has complete metadata.""" | |
| missing_fields = [] | |
| if not fact.tag: | |
| missing_fields.append("xbrl_tag") | |
| if not fact.period_end: | |
| missing_fields.append("period_end") | |
| if not fact.filed: | |
| missing_fields.append("filing_date") | |
| if not fact.form: | |
| missing_fields.append("form_type") | |
| if missing_fields: | |
| return ValidationResult( | |
| check_name="completeness_check", | |
| status="warning", | |
| message=f"Missing metadata: {', '.join(missing_fields)}", | |
| severity="medium" | |
| ) | |
| return ValidationResult( | |
| check_name="completeness_check", | |
| status="pass", | |
| message="Complete metadata", | |
| severity="low" | |
| ) | |
| def _calculate_confidence(self, fact, validation_results: List[ValidationResult]) -> int: | |
| """Calculate confidence score 0-100 based on validation results and source quality.""" | |
| base_confidence = 100 | |
| # Source quality component (40% of score) | |
| source_weight = self.source_weights.get(fact.form, 50) # default 50 for unknown forms | |
| source_component = int(source_weight * 0.4) | |
| # Data freshness component (20% of score) | |
| freshness_component = self._calculate_freshness_score(fact.filed) | |
| # Validation results component (40% of score) | |
| validation_component = self._calculate_validation_score(validation_results) | |
| total_confidence = source_component + freshness_component + validation_component | |
| return max(0, min(100, total_confidence)) # clamp to 0-100 | |
| def _calculate_freshness_score(self, filed_date: str) -> int: | |
| """Calculate freshness score based on filing date (20% of total confidence).""" | |
| if not filed_date: | |
| return 10 # low score for missing date | |
| try: | |
| filed = datetime.fromisoformat(filed_date.replace('Z', '+00:00')) | |
| days_old = (datetime.now().replace(tzinfo=filed.tzinfo) - filed).days | |
| if days_old <= 90: | |
| return 20 # fresh data | |
| elif days_old <= 365: | |
| return 15 # acceptable | |
| elif days_old <= 730: | |
| return 10 # getting stale | |
| else: | |
| return 5 # very stale | |
| except (ValueError, TypeError): | |
| return 10 # parsing error | |
| def _calculate_validation_score(self, validation_results: List[ValidationResult]) -> int: | |
| """Calculate validation score based on check results (40% of total confidence).""" | |
| total_weight = 0 | |
| weighted_score = 0 | |
| # Weight validation checks by importance | |
| check_weights = { | |
| 'magnitude_check': 15, # critical - catches data errors | |
| 'sign_check': 10, # important - catches obvious errors | |
| 'outlier_check': 10, # important - catches anomalies | |
| 'completeness_check': 5 # nice to have - metadata quality | |
| } | |
| for result in validation_results: | |
| weight = check_weights.get(result.check_name, 5) | |
| total_weight += weight | |
| if result.status == 'pass': | |
| weighted_score += weight | |
| elif result.status == 'warning': | |
| weighted_score += weight * 0.7 # partial credit | |
| else: # fail | |
| weighted_score += 0 # no credit | |
| if total_weight == 0: | |
| return 20 # default if no checks | |
| return int((weighted_score / total_weight) * 40) # scale to 40% of total confidence | |
| def _get_historical_context(self, ticker: str) -> Dict[str, List[float]]: | |
| """Get historical values for trend analysis.""" | |
| if not ticker: | |
| return {} | |
| try: | |
| from eugene.sources.xbrl import XBRLClient | |
| xbrl = XBRLClient(self.config) | |
| historical_data = {} | |
| key_metrics = ['revenue', 'operating_income', 'net_income', 'total_assets', | |
| 'total_debt', 'cash_and_equivalents'] | |
| for metric in key_metrics: | |
| try: | |
| historical_facts = xbrl.get_historical(ticker, metric, years=5) | |
| values = [fact.value for fact in historical_facts if fact.value is not None] | |
| if values: | |
| historical_data[metric] = values | |
| except Exception as e: | |
| logger.warning(f"Failed to get historical {metric} for {ticker}: {e}") | |
| return historical_data | |
| except Exception as e: | |
| logger.warning(f"Failed to get historical context for {ticker}: {e}") | |
| return {} | |
| def validate_xbrl_data(xbrl_financials, ticker=None, config=None): | |
| """ | |
| Convenience function to validate XBRL financial data. | |
| Args: | |
| xbrl_financials: XBRLFinancials object | |
| ticker: Optional ticker for historical comparison | |
| config: Optional config object | |
| Returns: | |
| Dict[str, ValidatedMetric]: Validated metrics | |
| """ | |
| validator = XBRLValidator(config) | |
| return validator.validate_financials(xbrl_financials, ticker) |