Aula 8: Regressão Descontínua e Synthetic Control

# Implementação de Regressão Descontínua em Python
import numpy as np
import pandas as pd
import statsmodels.formula.api as smf
import matplotlib.pyplot as plt
from rdrobust import rdrobust, rdplot

def rd_analysis(df, outcome_var, running_var, cutoff, bw=None, covariates=None):
    """
    Realiza análise de Regressão Descontínua
    
    Args:
        df: DataFrame com os dados
        outcome_var: Nome da variável de resultado
        running_var: Nome da variável de atribuição
        cutoff: Valor do limiar (cutoff)
        bw: Bandwidth (opcional, se None usa seleção ótima)
        covariates: Lista de covariáveis para controle (opcional)
Returns:
        Resultados da RD e gráficos
    """
    # Criar variável de tratamento
    df['treat'] = (df[running_var] >= cutoff).astype(int)
    
    # RD Estimação usando rdrobust (implementação robusta)
    if covariates:
        rd_results = rdrobust(y=df[outcome_var], x=df[running_var], 
                             c=cutoff, covs=df[covariates], h=bw)
    else:
        rd_results = rdrobust(y=df[outcome_var], x=df[running_var], 
                             c=cutoff, h=bw)
    
    # Criar gráfico de RD
    rdplot_results = rdplot(y=df[outcome_var], x=df[running_var], c=cutoff)
    
    # Dados para análise de placebo (teste de continuidade de covariáveis)
    placebo_tests = {}
    if covariates:
        for cov in covariates:
            placebo = rdrobust(y=df[cov], x=df[running_var], c=cutoff)
            placebo_tests[cov] = placebo.coef[0]
    
    return rd_results, rdplot_results, placebo_tests

# Implementação de Synthetic Control em Python
import pandas as pd
import numpy as np
from sklearn import linear_model
import matplotlib.pyplot as plt

def synthetic_control(df, unit_column, time_column, outcome_column, treated_unit,
                     treatment_period, predictors=None):
    """
    Implementa o método Synthetic Control
    
    Args:
        df: DataFrame com os dados em formato painel
        unit_column: Nome da coluna que identifica as unidades
        time_column: Nome da coluna de tempo
        outcome_column: Nome da coluna da variável de resultado
        treated_unit: Identificador da unidade tratada
        treatment_period: Período em que o tratamento ocorre
        predictors: Lista de colunas adicionais para usar como preditores além do outcome pré-tratamento
    
    Returns:
        Resultados do Synthetic Control e gráficos
    """
    # Separar períodos pré e pós tratamento
    pre_treatment = df[df[time_column] < treatment_period]
    post_treatment = df[df[time_column] >= treatment_period]
    
    # Identificar unidades de controle
    control_units = df[df[unit_column] != treated_unit][unit_column].unique()
    
    # Preparar dados para a unidade tratada
    treated_pre = pre_treatment[pre_treatment[unit_column] == treated_unit]
    treated_post = post_treatment[post_treatment[unit_column] == treated_unit]
    
    # Construir matriz de preditores para unidade tratada e controles
    X1 = treated_pre.pivot(index=time_column, columns=unit_column, values=outcome_column)
    X0 = pre_treatment[pre_treatment[unit_column].isin(control_units)].pivot(
        index=time_column, columns=unit_column, values=outcome_column)
    
    # Adicionar outros preditores, se especificados
    if predictors:
        for predictor in predictors:
            pred_treated = pre_treatment[pre_treatment[unit_column] == treated_unit][predictor].mean()
            pred_controls = pre_treatment[pre_treatment[unit_column].isin(control_units)].groupby(unit_column)[predictor].mean()
            
            # Adicionar às matrizes
            X1[predictor] = pred_treated
            for unit in control_units:
                X0[predictor] = pred_controls
    
    # Encontrar pesos ótimos usando o método de otimização
    synth_model = linear_model.Lasso(alpha=0.01, positive=True)
    synth_model.fit(X0.values, X1.iloc[:, 0].values)
    weights = synth_model.coef_ / np.sum(synth_model.coef_)
    
    # Construir controle sintético
    synthetic_pre = np.zeros(len(treated_pre))
    synthetic_post = np.zeros(len(treated_post))
    
    for i, unit in enumerate(control_units):
        unit_pre = pre_treatment[pre_treatment[unit_column] == unit][outcome_column].values
        unit_post = post_treatment[post_treatment[unit_column] == unit][outcome_column].values
        
        synthetic_pre += weights[i] * unit_pre
        synthetic_post += weights[i] * unit_post
    
    # Calcular efeito do tratamento
    treatment_effect = treated_post[outcome_column].values - synthetic_post
    
    # Preparar resultados para retorno
    results = {
        'weights': dict(zip(control_units, weights)),
        'treated_pre': treated_pre[outcome_column].values,
        'treated_post': treated_post[outcome_column].values,
        'synthetic_pre': synthetic_pre,
        'synthetic_post': synthetic_post,
        'treatment_effect': treatment_effect,
        'time_pre': treated_pre[time_column].values,
        'time_post': treated_post[time_column].values
    }
    
    return results