osint-investigation

1#!/usr/bin/env python32"""Permutation test for donation/contract timing correlation (stdlib-only).3 4For each (donor, vendor) pair, compute the mean number of days between each5donation and the nearest contract award. Then shuffle contract award dates6N times within the observation window and compute the same statistic. The7one-tailed p-value is the fraction of permutations whose mean is <= the8observed mean (smaller distance = tighter clustering).9 10Adapted from ShinMegamiBoson/OpenPlanter (MIT). Differences:11  - Pure stdlib (no pandas / numpy)12  - Domain-agnostic (no snow-vendor / CRITICAL-politician filter)13  - Configurable column names via flags14  - Optional --seed for reproducibility15"""16from __future__ import annotations17 18import argparse19import csv20import datetime as dt21import json22import random23import statistics24from collections import defaultdict25from pathlib import Path26 27_DATE_FORMATS = ("%Y-%m-%d", "%m/%d/%Y", "%Y/%m/%d", "%m-%d-%Y", "%Y%m%d")28 29 30def parse_date(raw: str) -> dt.date | None:31    if not raw:32        return None33    raw = raw.strip()34    for fmt in _DATE_FORMATS:35        try:36            return dt.datetime.strptime(raw, fmt).date()37        except ValueError:38            continue39    return None40 41 42def _read(path: str) -> list[dict[str, str]]:43    with open(path, newline="", encoding="utf-8") as fh:44        return list(csv.DictReader(fh))45 46 47def _nearest_distance(donation_date: dt.date, awards: list[dt.date]) -> int:48    """Absolute days to nearest award date."""49    return min(abs((donation_date - a).days) for a in awards)50 51 52def _permute(53    awards_count: int,54    donations: list[dt.date],55    date_min: dt.date,56    date_max: dt.date,57    rng: random.Random,58) -> float:59    """One permutation: draw uniform random award dates, compute mean nearest-distance."""60    span_days = (date_max - date_min).days or 161    rand_awards = [62        date_min + dt.timedelta(days=rng.randint(0, span_days))63        for _ in range(awards_count)64    ]65    distances = [_nearest_distance(d, rand_awards) for d in donations]66    return statistics.mean(distances)67 68 69def analyze(70    donations_path: str,71    donation_date_col: str,72    donation_amount_col: str,73    donation_donor_col: str,74    donation_recipient_col: str,75    contracts_path: str,76    contract_date_col: str,77    contract_vendor_col: str,78    cross_links_path: str | None,79    n_permutations: int = 1000,80    min_donations: int = 3,81    p_threshold: float = 0.05,82    seed: int | None = None,83    out_path: str = "timing.json",84) -> dict:85    rng = random.Random(seed)86 87    donations = _read(donations_path)88    contracts = _read(contracts_path)89 90    # Allow optional join through cross_links — donor (left) ↔ vendor (right).91    # When present, donor strings get mapped to matched vendor names so the92    # vendor-date index lookup actually finds the contracts.93    matched_pairs: set[tuple[str, str]] | None = None94    donor_to_vendors: dict[str, set[str]] = defaultdict(set)95    if cross_links_path:96        matched_pairs = set()97        for row in _read(cross_links_path):98            left = row.get("left_name", "")99            right = row.get("right_name", "")100            matched_pairs.add((left, right))101            donor_to_vendors[left].add(right)102 103    # Index contract dates by vendor name.104    vendor_to_award_dates: dict[str, list[dt.date]] = defaultdict(list)105    all_award_dates: list[dt.date] = []106    for row in contracts:107        d = parse_date(row.get(contract_date_col, ""))108        if not d:109            continue110        vendor_to_award_dates[row.get(contract_vendor_col, "").strip()].append(d)111        all_award_dates.append(d)112 113    if not all_award_dates:114        raise SystemExit(f"No parseable dates in {contracts_path}/{contract_date_col}")115    global_min = min(all_award_dates)116    global_max = max(all_award_dates)117 118    # Group donations by (donor, recipient).119    grouped: dict[tuple[str, str], list[tuple[dt.date, float]]] = defaultdict(list)120    for row in donations:121        donor = row.get(donation_donor_col, "").strip()122        recip = row.get(donation_recipient_col, "").strip()123        d = parse_date(row.get(donation_date_col, ""))124        try:125            amt = float(row.get(donation_amount_col, "0") or 0)126        except ValueError:127            amt = 0.0128        if not (donor and recip and d):129            continue130        grouped[(donor, recip)].append((d, amt))131 132    results = []133    skipped = 0134    for (donor, recip), records in grouped.items():135        if len(records) < min_donations:136            skipped += 1137            continue138        # Only test if donor appears in cross-links (when provided). The139        # (donor, candidate) tuple itself is NOT what's in matched_pairs —140        # cross_links pairs are (donor, vendor). We use the cross-link to141        # map donor → vendor name(s) so the vendor-date index resolves.142        if matched_pairs is not None and donor not in donor_to_vendors:143            skipped += 1144            continue145        # Try direct donor→awards first, then go through cross-link vendor names.146        award_dates = list(vendor_to_award_dates.get(donor, []))147        if not award_dates:148            award_dates = list(vendor_to_award_dates.get(recip, []))149        if not award_dates and donor_to_vendors.get(donor):150            for vendor_name in donor_to_vendors[donor]:151                award_dates.extend(vendor_to_award_dates.get(vendor_name, []))152        if not award_dates:153            skipped += 1154            continue155 156        donation_dates = [d for (d, _) in records]157        observed = statistics.mean(158            _nearest_distance(d, award_dates) for d in donation_dates159        )160 161        permuted_means = [162            _permute(len(award_dates), donation_dates, global_min, global_max, rng)163            for _ in range(n_permutations)164        ]165        p_value = sum(1 for m in permuted_means if m <= observed) / n_permutations166        null_mean = statistics.mean(permuted_means)167        null_std = statistics.pstdev(permuted_means) or 1.0168        effect_size = (null_mean - observed) / null_std169 170        results.append(171            {172                "donor": donor,173                "recipient": recip,174                "n_donations": len(records),175                "n_award_dates": len(award_dates),176                "observed_mean_days": round(observed, 2),177                "null_mean_days": round(null_mean, 2),178                "p_value": round(p_value, 4),179                "effect_size_sd": round(effect_size, 2),180                "significant": p_value < p_threshold,181                "total_donation_amount": round(sum(a for (_, a) in records), 2),182            }183        )184 185    results.sort(key=lambda r: r["p_value"])186 187    payload = {188        "metadata": {189            "n_permutations": n_permutations,190            "min_donations": min_donations,191            "p_threshold": p_threshold,192            "seed": seed,193            "n_pairs_tested": len(results),194            "n_pairs_skipped": skipped,195            "n_significant": sum(1 for r in results if r["significant"]),196            "observation_window": [global_min.isoformat(), global_max.isoformat()],197        },198        "results": results,199    }200 201    Path(out_path).write_text(json.dumps(payload, indent=2), encoding="utf-8")202    return payload203 204 205def main() -> int:206    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)207    p.add_argument("--donations", required=True)208    p.add_argument("--donation-date-col", required=True)209    p.add_argument("--donation-amount-col", required=True)210    p.add_argument("--donation-donor-col", required=True)211    p.add_argument("--donation-recipient-col", required=True)212    p.add_argument("--contracts", required=True)213    p.add_argument("--contract-date-col", required=True)214    p.add_argument("--contract-vendor-col", required=True)215    p.add_argument(216        "--cross-links",217        help="Optional cross_links.csv to restrict (donor, vendor) pairs",218    )219    p.add_argument("--permutations", type=int, default=1000)220    p.add_argument("--min-donations", type=int, default=3)221    p.add_argument("--p-threshold", type=float, default=0.05)222    p.add_argument("--seed", type=int)223    p.add_argument("--out", default="timing.json")224    a = p.parse_args()225 226    payload = analyze(227        donations_path=a.donations,228        donation_date_col=a.donation_date_col,229        donation_amount_col=a.donation_amount_col,230        donation_donor_col=a.donation_donor_col,231        donation_recipient_col=a.donation_recipient_col,232        contracts_path=a.contracts,233        contract_date_col=a.contract_date_col,234        contract_vendor_col=a.contract_vendor_col,235        cross_links_path=a.cross_links,236        n_permutations=a.permutations,237        min_donations=a.min_donations,238        p_threshold=a.p_threshold,239        seed=a.seed,240        out_path=a.out,241    )242    meta = payload["metadata"]243    print(244        f"Tested {meta['n_pairs_tested']} pairs ({meta['n_pairs_skipped']} skipped). "245        f"Significant (p<{meta['p_threshold']}): {meta['n_significant']}. "246        f"Wrote {a.out}"247    )248    return 0249 250 251if __name__ == "__main__":252    raise SystemExit(main())253