技能创建器
skill-creator
创建、改进和评估自定义技能,把专业流程转化为可重复使用的 AI 能力。
ds-observational-studies【数据与研究】"Analyzes observational data using matching methods, propensity scores stratification, and adjustment for confounding bias"
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: ds-observational-studies compatibility: opencode completeness: 95 content-types:
Comprehensive guide to observational studies in machine learning and data science workflows.
Observational Studies is a critical component of the machine learning workflow. This skill covers:
import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, roc_auc_score
def compute_propensity_scores(df: pd.DataFrame, treatment_col: str, covariates: list) -> dict:
"""Compute propensity scores and evaluate model performance."""
if treatment_col not in df.columns:
raise ValueError(f"Treatment column '{treatment_col}' not found in DataFrame.")
X = df[covariates].dropna()
y = df.loc[X.index, treatment_col]
model = LogisticRegression(max_iter=1000, random_state=42)
model.fit(X, y)
scores = pd.Series(model.predict_proba(X)[:, 1], index=X.index, name='propensity_score')
y_pred = model.predict(X)
metrics = {
'accuracy': accuracy_score(y, y_pred)
'roc_auc': roc_auc_score(y, scores)
}
return {'propensity_scores': scores, 'model_metrics': metrics}
# Example usage with synthetic data
if __name__ == "__main__":
np.random.seed(42)
n_samples = 500
data = pd.DataFrame({
'age': np.random.normal(50, 10, n_samples)
'income': np.random.normal(60000, 15000, n_samples)
'treatment': np.random.binomial(1, 0.5, n_samples)
})
covariates = ['age', 'income']
result = compute_propensity_scores(data, 'treatment', covariates)
print(f"Computed scores for {len(result['propensity_scores'])} samples")
print(f"Model Metrics: {result['model_metrics']}")
import logging
from typing import Any, Dict, List
import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression
from sklearn.neighbors import NearestNeighbors
logger = logging.getLogger(__name__)
class ObservationalStudyPipeline:
"""Production implementation for observational study analysis."""
def __init__(self, propensity_model=None, matching_method: str = 'nearest_neighbor'):
self.propensity_model = propensity_model or LogisticRegression(max_iter=1000, random_state=42)
self.matching_method = matching_method
self.results: Dict[str, Any] = {}
def _compute_propensity(self, df: pd.DataFrame, treatment_col: str, covariates: List[str]) -> pd.Series:
X = df[covariates].dropna()
y = df.loc[X.index, treatment_col]
self.propensity_model.fit(X, y)
return pd.Series(self.propensity_model.predict_proba(X)[:, 1], index=X.index, name='propensity_score')
def _match_treatments(self, df: pd.DataFrame, caliper: float = 0.05) -> pd.DataFrame:
treated = df[df['treatment'] == 1].dropna(subset=['propensity_score'])
control = df[df['treatment'] == 0].dropna(subset=['propensity_score'])
if len(treated) == 0 or len(control) == 0:
raise ValueError("No treated or control observations found after dropping NaNs.")
nn = NearestNeighbors(n_neighbors=1)
nn.fit(control[['propensity_score']])
distances, indices = nn.kneighbors(treated[['propensity_score']])
mask = distances.flatten() < caliper
matched_control = control.iloc[indices.flatten()[mask]]
matched_treated = treated.iloc[mask]
return pd.concat([matched_treated, matched_control], ignore_index=True)
def execute(self, data: pd.DataFrame, treatment_col: str, covariates: List[str]) -> Dict[str, Any]:
logger.info("Starting observational study pipeline")
if data.empty:
raise ValueError("Input data cannot be empty")
df = data.copy()
df['propensity_score'] = self._compute_propensity(df, treatment_col, covariates)
matched_df = self._match_treatments(df)
self.results = {
'status': 'success'
'matched_samples': len(matched_df)
'treated_count': len(matched_df[matched_df['treatment'] == 1])
'control_count': len(matched_df[matched_df['treatment'] == 0])
'matched_data': matched_df
}
logger.info(f"Pipeline completed. Matched {self.results['matched_samples']} samples.")
return self.results
| Pitfall | Problem | Solution | |
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.