Spaces:
Sleeping
Sleeping
| import logging | |
| import pandas as pd | |
| import numpy as np | |
| from typing import Optional | |
| from core.memory_guard import force_gc, optimize_dataframe | |
| logger = logging.getLogger("nexus.features") | |
| class NexusFeatureEngine: | |
| def __init__(self, loader=None): | |
| self.loader = loader | |
| def generate(self, df): | |
| """Generate features for a DataFrame of matches.""" | |
| if df.empty: | |
| return df | |
| features = df.copy() | |
| # Quick return without DB context | |
| if not self.loader: | |
| features = self._add_basic_features(features) | |
| logger.info(f"Generated shallow features: {features.shape}") | |
| return features | |
| # 0. Advanced Statistical Models (Extra Power) | |
| try: | |
| from features.statistical import StatisticalFeatureEngine | |
| # We need historical context for these models | |
| # Fetch last results for context | |
| hist_query = """ | |
| SELECT match_id, home_team_id, away_team_id, home_goals, away_goals, kickoff | |
| FROM results | |
| WHERE kickoff < (SELECT MIN(kickoff) FROM df) | |
| AND (home_goals IS NOT NULL AND away_goals IS NOT NULL) | |
| ORDER BY kickoff DESC | |
| LIMIT 250000 | |
| """ | |
| self.loader.con.register("input_features", features) | |
| hist_df = self.loader.con.execute(hist_query.replace("df", "input_features")).fetchdf() | |
| sfe = StatisticalFeatureEngine() | |
| # SCENARIO: Calibrating on full history (df includes all past data) | |
| # If history is empty but df is large, use df as the training set for statistical priors | |
| if hist_df.empty and len(df) > 10000: | |
| logger.info("Retraining scenario detected. Using full dataframe for statistical calibration.") | |
| features = sfe.generate_features(features, df) | |
| else: | |
| features = sfe.generate_features(features, hist_df) | |
| logger.info(f"Successfully integrated Extra Power statistical priors. Shape now: {features.shape}") | |
| features = optimize_dataframe(features) | |
| force_gc() | |
| except Exception as e: | |
| logger.error(f"Failed to integrate Extra Power features: {e}", exc_info=True) | |
| # 1. Classical rating algorithms | |
| from features.classical import RatingEngine | |
| re = RatingEngine() | |
| features = re.calculate_elo(features) | |
| features = re.calculate_pi(features) | |
| # 2. Add DuckDB advanced rolling features | |
| from features.advanced import AdvancedFeatureEngine | |
| try: | |
| advanced_engine = AdvancedFeatureEngine(self.loader.con) | |
| adv_df = advanced_engine.generate_rolling_features() | |
| # Merge advanced features into dataframe - drop existing to avoid suffix collisions | |
| cols_to_drop = [c for c in adv_df.columns if c in features.columns and c != 'match_id'] | |
| if cols_to_drop: | |
| features = features.drop(columns=cols_to_drop) | |
| features = pd.merge(features, adv_df, on='match_id', how='left') | |
| logger.info("Successfully merged advanced rolling features") | |
| features = optimize_dataframe(features) | |
| force_gc() | |
| except Exception as e: | |
| logger.error(f"Failed to generate advanced features: {e}") | |
| # 3. Add Contextual & Behavioral Features (ACTIVATED Phase v4) | |
| try: | |
| from features.contextual import ContextualIntelligence | |
| context_engine = ContextualIntelligence() | |
| features = context_engine.apply_all(features) | |
| logger.info(f"Contextual Intelligence applied. Shape: {features.shape}") | |
| features = optimize_dataframe(features) | |
| except Exception as e: | |
| logger.warning(f"Contextual feature generation bypassed or failed: {e}") | |
| # 4. Add Player Impact Features (LSI & Absences) | |
| try: | |
| from features.player_metrics import PlayerImpactEngine | |
| player_engine = PlayerImpactEngine() | |
| features = player_engine.generate_player_features(features) | |
| logger.info("Successfully merged Player Impact and LSI features") | |
| except Exception as e: | |
| logger.error(f"Failed to generate Player Metrics features: {e}") | |
| # 5. Add Market Efficiency Features (True Odds & CLV) | |
| try: | |
| from features.market_implied import MarketEfficiencyEngine | |
| market_engine = MarketEfficiencyEngine() | |
| market_df = market_engine.generate_market_features(features) | |
| # Collision-safe merge | |
| cols_to_drop = [c for c in market_df.columns if c in features.columns and c != 'match_id'] | |
| if cols_to_drop: | |
| features = features.drop(columns=cols_to_drop) | |
| features = pd.merge(features, market_df, on='match_id', how='left') | |
| logger.info("Successfully merged Market Implied features") | |
| features = optimize_dataframe(features) | |
| force_gc() | |
| except Exception as e: | |
| logger.error(f"Failed to generate Market Implied features: {e}") | |
| # 6. Apply Synthetic Data Augmentation (Conditional) | |
| try: | |
| from features.gan_augmentation import SyntheticDataAugmentor | |
| # we only run this on historical datasets, not on live inference | |
| if len(features) > 100: | |
| augmentor = SyntheticDataAugmentor(target_ratio=0.8) # Bring draws/away close to home ratio | |
| logger.info("Synthetic Generative bindings active") | |
| # actual augmentation runs in data_loader prior to model hand-off | |
| except Exception as e: | |
| logger.warning(f"GAN Augmentation bypassed or failed: {e}") | |
| # [PHASE 13] Tactical Intelligence (xT & Style DNA) | |
| try: | |
| features = self._add_tactical_features(features) | |
| logger.info("Successfully merged Tactical Intelligence (xT) features") | |
| except Exception as e: | |
| logger.error(f"Failed to generate Tactical features: {e}") | |
| # 7. Handle default fillings & hydration | |
| features = self._hydrate_latest_ratings(features) | |
| features = self._add_basic_features(features) | |
| logger.info(f"Generated complete feature stack: {features.shape}") | |
| return features | |
| def _add_tactical_features(self, df): | |
| """Phase 13 & 15: Convert raw data into Tactical Identity & Qualitative Residuals.""" | |
| from features.tactical import ExpectedThreatEngine, StyleDNA | |
| from features.contextual_llm import QualitativeObserver | |
| xt_engine = ExpectedThreatEngine() | |
| qual_observer = QualitativeObserver(llm_gateway=getattr(self, 'llm', None)) | |
| # 1. Expected Threat (xT) | |
| if 'incidents' in df.columns: | |
| def _calc_row_xt(row): | |
| incidents = row.get('incidents') | |
| # Explicit check to avoid "truth value of an array is ambiguous" | |
| if incidents is None or not hasattr(incidents, '__len__') or len(incidents) == 0: | |
| return pd.Series([0.0, 0.0, 0.0]) | |
| res = xt_engine.calculate_match_xt(incidents) | |
| return pd.Series([res['home_xt_total'], res['away_xt_total'], res['net_xt_margin']]) | |
| xt_cols = df.apply(_calc_row_xt, axis=1) | |
| xt_cols.columns = ['home_xt_total', 'away_xt_total', 'net_xt_margin'] | |
| # Collision-safe concat | |
| for col in xt_cols.columns: | |
| if col in df.columns: | |
| df = df.drop(columns=[col]) | |
| df = pd.concat([df, xt_cols], axis=1) | |
| else: | |
| df['home_xt_total'] = 0.0 | |
| df['away_xt_total'] = 0.0 | |
| df['net_xt_margin'] = 0.0 | |
| # 2. Style DNA (Phase 13.2) | |
| if self.loader and self.loader.con: | |
| try: | |
| dna_engine = StyleDNA(con=self.loader.con) | |
| dna_df = dna_engine.generate_style_features() | |
| if not dna_df.empty: | |
| # Drop existing DNA columns to avoid suffix collisions | |
| cols_to_drop = [c for c in dna_df.columns if c in df.columns and c != 'match_id'] | |
| if cols_to_drop: | |
| df = df.drop(columns=cols_to_drop) | |
| df = pd.merge(df, dna_df, on='match_id', how='left') | |
| logger.info(f"Successfully integrated Tactical Identity vectors") | |
| except Exception as e: | |
| logger.warning(f"Style DNA extraction skipped or failed: {e}") | |
| # 3. Qualitative Residuals (Phase 15) | |
| try: | |
| df = qual_observer.add_qualitative_features(df) | |
| logger.info("Qualitative Motivation residuals active") | |
| except Exception as e: | |
| logger.warning(f"Qualitative extraction failed: {e}") | |
| return df | |
| def _hydrate_latest_ratings(self, df): | |
| """ | |
| Lookup team ratings as they stood AT THE TIME OF KICKOFF. | |
| This is critical to avoid Look-Ahead Bias / Data Leakage. | |
| """ | |
| if df.empty or 'kickoff' not in df.columns: | |
| return df | |
| try: | |
| # We use a correlated subquery or a window function join in DuckDB to find the | |
| # most recent ratings that existed strictly BEFORE each match's kickoff. | |
| # Convert DF to a temporary table for the join | |
| self.loader.con.register("df_to_hydrate", df) | |
| self.loader.con.execute("CREATE OR REPLACE TEMPORARY TABLE matches_to_hydrate AS SELECT * FROM df_to_hydrate") | |
| results = self.loader.con.execute(""" | |
| WITH team_ratings AS ( | |
| SELECT match_id, home_team_id as team_id, home_elo as elo, home_pi as pi, kickoff | |
| FROM nexus_features WHERE home_elo IS NOT NULL | |
| UNION ALL | |
| SELECT match_id, away_team_id as team_id, away_elo as elo, away_pi as pi, kickoff | |
| FROM nexus_features WHERE away_elo IS NOT NULL | |
| ), | |
| home_hydrated AS ( | |
| SELECT | |
| m.match_id, | |
| (SELECT elo FROM team_ratings tr | |
| WHERE tr.team_id = m.home_team_id AND tr.kickoff < m.kickoff | |
| ORDER BY tr.kickoff DESC LIMIT 1) as home_elo_hist, | |
| (SELECT pi FROM team_ratings tr | |
| WHERE tr.team_id = m.home_team_id AND tr.kickoff < m.kickoff | |
| ORDER BY tr.kickoff DESC LIMIT 1) as home_pi_hist | |
| FROM matches_to_hydrate m | |
| ), | |
| away_hydrated AS ( | |
| SELECT | |
| m.match_id, | |
| (SELECT elo FROM team_ratings tr | |
| WHERE tr.team_id = m.away_team_id AND tr.kickoff < m.kickoff | |
| ORDER BY tr.kickoff DESC LIMIT 1) as away_elo_hist, | |
| (SELECT pi FROM team_ratings tr | |
| WHERE tr.team_id = m.away_team_id AND tr.kickoff < m.kickoff | |
| ORDER BY tr.kickoff DESC LIMIT 1) as away_pi_hist | |
| FROM matches_to_hydrate m | |
| ) | |
| SELECT | |
| m.match_id, | |
| COALESCE(h.home_elo_hist, 1500.0) as home_elo, | |
| COALESCE(h.home_pi_hist, 0.0) as home_pi, | |
| COALESCE(a.away_elo_hist, 1500.0) as away_elo, | |
| COALESCE(a.away_pi_hist, 0.0) as away_pi | |
| FROM matches_to_hydrate m | |
| JOIN home_hydrated h ON m.match_id = h.match_id | |
| JOIN away_hydrated a ON m.match_id = a.match_id | |
| """).fetchdf() | |
| # Merge hydrated columns back into the original df | |
| # Drop existing elo/pi columns if they exist to avoid suffixes | |
| cols_to_drop = [c for c in results.columns if c in df.columns and c != 'match_id'] | |
| if cols_to_drop: | |
| df = df.drop(columns=cols_to_drop) | |
| df = pd.merge(df, results, on='match_id', how='left') | |
| # Recalculate diffs | |
| df['elo_diff'] = df['home_elo'] - df['away_elo'] | |
| df['pi_diff'] = df['home_pi'] - df['away_pi'] | |
| logger.info(f"Point-in-time hydration complete for {len(df)} matches (Zero Leakage)") | |
| return df | |
| except Exception as e: | |
| logger.warning(f"Failed point-in-time hydration: {e}. Falling back to default.") | |
| return df | |
| def _add_basic_features(self, df): | |
| """Add basic features - fill missing ratings and calculate stats.""" | |
| cols = ["home_elo", "away_elo", "elo_diff", "home_form", "away_form", | |
| "home_pi", "away_pi", "pi_diff", "home_xg", "away_xg"] | |
| for col in cols: | |
| if col not in df.columns: | |
| if col in ["home_elo", "away_elo"]: | |
| df[col] = 1500.0 | |
| else: | |
| df[col] = 0.0 | |
| else: | |
| if col in ["home_elo", "away_elo"]: | |
| df[col] = df[col].fillna(1500.0) | |
| else: | |
| df[col] = df[col].fillna(0.0) | |
| # Recalculate diffs after filling NAs to ensure consistency | |
| if 'home_elo' in df.columns and 'away_elo' in df.columns: | |
| df['elo_diff'] = df['home_elo'] - df['away_elo'] | |
| if 'home_pi' in df.columns and 'away_pi' in df.columns: | |
| df['pi_diff'] = df['home_pi'] - df['away_pi'] | |
| return df | |