Rct Randomization

Randomization must be reproducible, documented, and executed exactly once per study.

import numpy as np
import pandas as pd
from datetime import datetime

# Set seed ONCE at beginning (DOCUMENT THIS VALUE)
RANDOM_SEED = 42
np.random.seed(RANDOM_SEED)

# Load eligible sample
df = pd.read_csv("/mnt/data/eligible_units.csv")
n = len(df)

print(f"RANDOMIZATION PROTOCOL")
print(f"Date: {datetime.now()}")
print(f"Random seed: {RANDOM_SEED}")
print(f"Sample size: {n}")

# Stratified permutation randomization
treatment_prop = 0.5
strata_var = 'baseline_risk'  # Strong outcome predictor

df['treatment'] = 0

for stratum in df[strata_var].unique():
    strata_mask = df[strata_var] == stratum
    n_strata = strata_mask.sum()
    n_treat = int(n_strata * treatment_prop)

    # Permutation within stratum
    assignments = np.concatenate([
        np.ones(n_treat, dtype=int),
        np.zeros(n_strata - n_treat, dtype=int)
    ])
    np.random.shuffle(assignments)

    df.loc[strata_mask, 'treatment'] = assignments

# CRITICAL: Save immediately (NEVER re-run randomization)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
df.to_csv(f"/mnt/data/randomization_assignments_{timestamp}.csv", index=False)

# Create backup
df.to_csv(f"/mnt/data/randomization_assignments_BACKUP.csv", index=False)

print(f"\nAssignments saved to:")
print(f"  /mnt/data/randomization_assignments_{timestamp}.csv")
print(f"Treatment: {df['treatment'].sum()} ({df['treatment'].mean():.1%})")
print(f"Control: {(~df['treatment'].astype(bool)).sum()} ({(~df['treatment'].astype(bool)).mean():.1%})")

return df,

</code_template>

</implementation_pattern>

<examples>

<example context="cluster_randomization" difficulty="intermediate">
<description>Cluster randomization to prevent spillovers</description>
<code>
```python
@app.cell
def cluster_randomize(df):
    # Cluster randomization when spillovers likely within clusters
    # Reduces power due to design effect, but necessary for validity

    import numpy as np
    import pandas as pd

    # CRITICAL: Set seed once
    np.random.seed(123)

    # Identify clusters
    clusters = df['village_id'].unique()
    n_clusters = len(clusters)
    n_treat_clusters = int(n_clusters * 0.5)

    print(f"CLUSTER RANDOMIZATION")
    print(f"Total clusters: {n_clusters}")
    print(f"Treatment clusters: {n_treat_clusters}")

    # Permutation at cluster level
    cluster_assignments = np.concatenate([
        np.ones(n_treat_clusters, dtype=int),
        np.zeros(n_clusters - n_treat_clusters, dtype=int)
    ])
    np.random.shuffle(cluster_assignments)

    # Map cluster assignments to individuals
    cluster_map = dict(zip(clusters, cluster_assignments))
    df['treatment'] = df['village_id'].map(cluster_map)

    # Report
    print(f"\nIndividuals in treatment: {df['treatment'].sum()}")
    print(f"Individuals in control: {(~df['treatment'].astype(bool)).sum()}")

    # Calculate design effect
    avg_cluster_size = df.groupby('village_id').size().mean()
    icc = 0.05  # Assumed intra-cluster correlation
    design_effect = 1 + (avg_cluster_size - 1) * icc

    print(f"\nAverage cluster size: {avg_cluster_size:.1f}")
    print(f"Assumed ICC: {icc:.3f}")
    print(f"Design effect: {design_effect:.2f}")
    print(f"Effective sample size: {len(df)/design_effect:.0f}")

    return df,

Never re-randomize based on balance. If you must, use proper re-randomization protocol with inference adjustment (complex).