python案例如何应对小联赛数据缺失问题？

wen python案例 2026-08-06 166

本文目录导读：

数据获取与扩充策略
数据预处理与特征工程
模型构建与自适应策略
预测结果的置信度评估
实战案例：小联赛比分预测
监控与持续改进

针对小联赛数据缺失的问题,Python可以通过多种策略来应对，以下是系统性的解决方案，从数据获取到模型构建的完整流程：

数据获取与扩充策略

多源数据整合

import requests
import pandas as pd
from bs4 import BeautifulSoup
def fetch_multiple_sources(league_name):
    """从多个数据源获取小联赛数据"""
    sources = [
        {'name': 'FootyStats', 'url': f'https://footystats.org/clubs/{league_name}'},
        {'name': 'SofaScore', 'url': f'https://www.sofascore.com/team/football/{league_name}'},
        {'name': 'FlashScore', 'url': f'https://www.flashscore.com/football/{league_name}/'}
    ]
    data_frames = []
    for source in sources:
        try:
            # 模拟请求，实际需要处理反爬
            response = requests.get(source['url'], headers={'User-Agent': 'Mozilla/5.0'})
            soup = BeautifulSoup(response.content, 'html.parser')
            # 提取数据逻辑（根据具体网站结构调整）
            df = extract_data_from_soup(soup)
            data_frames.append(df)
        except Exception as e:
            print(f"从{source['name']}获取失败: {e}")
    # 合并多个源的数据
    combined_df = pd.concat(data_frames, ignore_index=True)
    return combined_df.drop_duplicates(subset=['match_id'])

官方API和爬虫结合

def collect_official_data():
    """利用官方API或赛事网站获取基础数据"""
    # 例如FIFA排名API
    fifa_api = "https://api.fifa.com/api/v3/rankings"
    # 对于小联赛，可以考虑：
    # - 各国足协官网
    # - 联赛官方网站
    # - 俱乐部官方社交媒体（球员伤病、状态）
    # 结合爬虫获取比赛报道中的统计信息
    match_reports = extract_match_reports_from_news()
    return match_reports

数据预处理与特征工程

处理缺失值的高级方法

import numpy as np
from sklearn.impute import KNNImputer
from sklearn.ensemble import RandomForestRegressor
def advanced_missing_value_handling(df):
    """多策略缺失值处理"""
    # 1. 基于时间序列的插值
    for col in df.columns:
        if col not in ['date', 'match_id']:
            # 使用前向填充和后向填充的组合
            df[col] = df[col].fillna(method='ffill').fillna(method='bfill')
    # 2. 基于相关性的KNN插补
    imputer = KNNImputer(n_neighbors=5, weights='distance')
    df_imputed = pd.DataFrame(
        imputer.fit_transform(df.select_dtypes(include=[np.number])),
        columns=df.select_dtypes(include=[np.number]).columns
    )
    # 3. 基于机器学习模型预测缺失值
    for col in df.columns:
        if df[col].isna().sum() > 0:
            # 构建随机森林回归模型预测
            col_without_na = df[df[col].notna()]
            X = col_without_na.drop(columns=[col])
            y = col_without_na[col]
            model = RandomForestRegressor(n_estimators=100)
            model.fit(X, y)
            # 预测缺失值
            df.loc[df[col].isna(), col] = model.predict(
                df[df[col].isna()].drop(columns=[col])
            )
    return df

特征工程增强

def create_advanced_features(df):
    """创建高级特征以补偿数据缺失"""
    # 基于联赛级别的统计特征
    df['league_avg_goals'] = df.groupby('league')['goals_scored'].transform('mean')
    df['league_avg_conceded'] = df.groupby('league')['goals_conceded'].transform('mean')
    # 时间衰减特征（近期状态权重更高）
    df['days_since_last_match'] = (df['date'] - df.groupby('team')['date'].shift()).dt.days
    # 滚动统计特征（适用于不完整的数据）
    for window in [5, 10, 15]:
        df[f'rolling_avg_goals_{window}'] = df.groupby('team')['goals_scored'].transform(
            lambda x: x.rolling(window, min_periods=1).mean()
        )
    # 对阵双方的历史交战记录（即使数据不完整）
    df['head_to_head_goals'] = df.groupby(['home_team', 'away_team'])['goals_scored'].transform('mean')
    return df

模型构建与自适应策略

集成学习处理数据稀疏性

from sklearn.ensemble import StackingClassifier, RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score
from sklearn.preprocessing import StandardScaler
def build_adaptive_model(X_train, y_train, X_test, missing_ratio):
    """构建自适应模型，根据数据完整度调整策略"""
    # 根据缺失比例调整模型复杂度
    if missing_ratio > 0.3:
        # 高缺失：使用简单模型 + 交叉验证
        models = [
            ('lr', LogisticRegression(max_iter=1000)),
            ('rf_simple', RandomForestClassifier(n_estimators=50, max_depth=3))
        ]
    else:
        # 低缺失：使用复杂模型
        models = [
            ('gb', GradientBoostingClassifier()),
            ('rf', RandomForestClassifier(n_estimators=200))
        ]
    # 堆叠集成
    stacking_model = StackingClassifier(
        estimators=models,
        final_estimator=LogisticRegression()
    )
    # 使用交叉验证评估
    scores = cross_val_score(stacking_model, X_train, y_train, cv=5)
    print(f"交叉验证得分: {scores.mean():.3f} (+/- {scores.std()*2:.3f})")
    return stacking_model

贝叶斯优化与不确定性处理

from scipy.stats import uniform, randint
from sklearn.model_selection import RandomizedSearchCV
import optuna
def bayesian_hyperparameter_tuning(model, X_train, y_train):
    """使用贝叶斯优化寻找最佳超参数"""
    def objective(trial):
        params = {
            'n_estimators': trial.suggest_int('n_estimators', 50, 300),
            'max_depth': trial.suggest_int('max_depth', 2, 10),
            'min_samples_split': trial.suggest_int('min_samples_split', 2, 20),
            'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3)
        }
        model.set_params(**params)
        scores = cross_val_score(model, X_train, y_train, cv=5)
        return scores.mean()
    study = optuna.create_study(direction='maximize')
    study.optimize(objective, n_trials=100)
    return study.best_params

预测结果的置信度评估

def prediction_with_confidence(model, X_test, n_iterations=100):
    """多次预测以获得置信度区间"""
    predictions = []
    for _ in range(n_iterations):
        # 模拟数据不确定性（小联赛数据特征）
        noise = np.random.normal(0, 0.1, X_test.shape)
        noisy_test = X_test + noise
        predictions.append(model.predict_proba(noisy_test))
    mean_pred = np.mean(predictions, axis=0)
    std_pred = np.std(predictions, axis=0)
    # 返回预测和置信度
    return mean_pred, std_pred

实战案例：小联赛比分预测

def predict_small_league_match(home_team, away_team, historical_data):
    """完整的小联赛比分预测流程"""
    # 1. 数据准备
    df = prepare_match_data(home_team, away_team, historical_data)
    # 2. 缺失值处理
    df = advanced_missing_value_handling(df)
    # 3. 特征工程
    df = create_advanced_features(df)
    # 4. 特征选择与模型训练
    features = ['home_win_ratio', 'away_win_ratio', 'avg_goals', 'recent_form', 
                'head_to_head', 'league_avg_goals', 'days_since_last_match']
    X = df[features]
    y = df['result']  # 0=主胜, 1=平, 2=客胜
    # 5. 模型训练与预测
    model = build_adaptive_model(X, y, X_test, missing_ratio=df.isna().sum().sum()/df.size)
    # 6. 输出预测结果及置信度
    probabilities = model.predict_proba(X_test)[0]
    confidence = max(probabilities) * 100
    return {
        'home': f"{home_team}胜", 
        'draw': '平局',
        'away': f"{away_team}胜",
        'probabilities': probabilities,
        'confidence': f"{confidence:.1f}%"
    }

监控与持续改进

class SmallLeagueModelMonitor:
    def __init__(self):
        self.predictions = []
        self.actuals = []
    def log_prediction(self, prediction, actual):
        """记录预测和实际结果用于评估"""
        self.predictions.append(prediction)
        self.actuals.append(actual)
    def performance_metrics(self):
        """计算模型性能指标"""
        from sklearn.metrics import accuracy_score, brier_score_loss
        accuracy = accuracy_score(self.actuals, self.predictions)
        brier = brier_score_loss(self.actuals, self.predictions)
        # 小联赛中，关注：
        # - 盈利显著性
        # - 预测稳定性
        # - 不同联赛的适应能力
        return {
            'accuracy': accuracy,
            'brier_score': brier,
            'sample_size': len(self.predictions)
        }
    def adapt_to_new_season(self, new_data):
        """新赛季数据来临时自动调整"""
        # 检测数据分布变化
        from scipy import stats
        kstest = stats.ks_2samp(self.actuals, new_data)
        if kstest.pvalue < 0.05:
            print("数据分布发生显著变化，需要重新训练模型")
            # 触发模型重训练逻辑

核心建议：