Skip to content

positive_control

Positive-control check: free-flow speed (a road-design characteristic) as a spatial predictor of current speed. Recovering this known spatial signal shows the pipeline can detect spatial structure where it exists.

positive_control

Positive-control analysis: free-flow speed as a spatial predictor of current speed.

Free-flow speed is a road-design characteristic (lane count, speed limit, road class) and should be strongly predictable from segment identity. Recovering this known spatial signal demonstrates that the pipeline can detect spatial structure where it exists, so a null centrality--congestion result cannot be attributed to insufficient power, spatial-matching errors, or data-quality problems.

Computes the Pearson R^2 between free-flow speed and current speed per city at two aggregation levels:

  • segment-level -- segment means collapsed across the temporal periods (the between-segment statistic reported in the manuscript)
  • pooled panel -- all segment x period observations

Outputs positive_control_r2.csv in the analysis-results directory.

load_panel 

load_panel(city_code, base_dir='.')

Build a long panel (segment x period) of current speed and free-flow speed.

Source code in src/trafficpipeline/positive_control.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def load_panel(city_code: str, base_dir: str | Path = ".") -> pd.DataFrame:
    """Build a long panel (segment x period) of current speed and free-flow speed."""
    folder = Path(base_dir) / CITIES[city_code]["traffic_output_dir"]
    frames = []
    for period in TIME_PERIODS:
        fp = folder / f"{period}_{city_code}.gpkg"
        if not fp.exists():
            continue
        gdf = gpd.read_file(str(fp))
        if "speed_mean" not in gdf.columns or "free_flow_mean" not in gdf.columns:
            continue
        frames.append(
            pd.DataFrame(
                {
                    "segment_id": gdf["osm_composite_id"],
                    "speed": gdf["speed_mean"],
                    "ffs": gdf["free_flow_mean"],
                }
            )
        )
    if not frames:
        raise FileNotFoundError(f"No period GPKGs with speed columns for {city_code}")
    return pd.concat(frames, ignore_index=True).dropna(subset=["speed", "ffs"])

compute_positive_control 

compute_positive_control(base_dir='.')

Per-city Pearson R^2 of free-flow speed vs current speed, both levels.

Source code in src/trafficpipeline/positive_control.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def compute_positive_control(base_dir: str | Path = ".") -> pd.DataFrame:
    """Per-city Pearson R^2 of free-flow speed vs current speed, both levels."""
    rows = []
    for code in CITIES:
        panel = load_panel(code, base_dir)
        seg = panel.groupby("segment_id")[["speed", "ffs"]].mean()

        r_seg, p_seg = stats.pearsonr(seg["ffs"], seg["speed"])
        r_pool, p_pool = stats.pearsonr(panel["ffs"], panel["speed"])

        rows.append(
            {
                "city": code,
                "n_segments": len(seg),
                "n_panel_obs": len(panel),
                "r2_segment_level": round(r_seg**2, 4),
                "p_segment_level": p_seg,
                "r2_pooled_panel": round(r_pool**2, 4),
                "p_pooled_panel": p_pool,
            }
        )
    return pd.DataFrame(rows)

run_analysis 

run_analysis(base_dir='.', output_dir='analysis_results')

Run the positive-control analysis and write positive_control_r2.csv.

Source code in src/trafficpipeline/positive_control.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def run_analysis(
    base_dir: str | Path = ".",
    output_dir: str | Path = "analysis_results",
) -> pd.DataFrame:
    """Run the positive-control analysis and write ``positive_control_r2.csv``."""
    df = compute_positive_control(base_dir)

    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    csv_path = out / "positive_control_r2.csv"
    df.to_csv(csv_path, index=False)

    for row in df.itertuples(index=False):
        print(
            f"{row.city}: segments={row.n_segments:,}  "
            f"segment-level R2={row.r2_segment_level:.3f}  "
            f"pooled R2={row.r2_pooled_panel:.3f}"
        )
    print(f"written: {csv_path}")
    return df