55 lines
2.0 KiB
Python
55 lines
2.0 KiB
Python
|
"""
|
||
|
|
core/regression_auditor.py
|
||
|
|
Automated rank shift regression tracking against historic snapshots.
|
||
|
|
Generates an audit report of movements exceeding the 100-position threshold.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
import pandas as pd
|
||
|
|
|
||
|
|
|
||
|
|
def audit_ranking_regression(
|
||
|
|
current_csv_path: str = "data/hiffi_multiplatform_500.csv",
|
||
|
|
snapshot_dir: str = "data/snapshots",
|
||
|
|
threshold: int = 100
|
||
|
|
) -> pd.DataFrame:
|
||
|
|
cur_path = Path(current_csv_path)
|
||
|
|
snap_path = Path(snapshot_dir)
|
||
|
|
snap_path.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
df_current = pd.read_csv(cur_path)
|
||
|
|
|
||
|
|
# Find most recent previous snapshot
|
||
|
|
snapshots = sorted(snap_path.glob("hiffi_ranking_*.csv"))
|
||
|
|
|
||
|
|
if not snapshots:
|
||
|
|
# Save initial baseline snapshot if none exists
|
||
|
|
today_str = datetime.now(timezone.utc).strftime("%Y%m%d")
|
||
|
|
initial_snap = snap_path / f"hiffi_ranking_{today_str}.csv"
|
||
|
|
df_current.to_csv(initial_snap, index=False)
|
||
|
|
print(f"[RegressionAuditor] Initial snapshot saved to {initial_snap}. No prior history to compare.")
|
||
|
|
return pd.DataFrame()
|
||
|
|
|
||
|
|
latest_snapshot = snapshots[-1]
|
||
|
|
print(f"[RegressionAuditor] Comparing current dataset against snapshot: {latest_snapshot.name}")
|
||
|
|
df_prev = pd.read_csv(latest_snapshot)
|
||
|
|
|
||
|
|
# Compare published ranks
|
||
|
|
cur_ranks = df_current.set_index("name")["hiffi_xp_published_rank"]
|
||
|
|
prev_ranks = df_prev.set_index("name")["hiffi_xp_published_rank"]
|
||
|
|
|
||
|
|
comparison = pd.DataFrame({
|
||
|
|
"previous_published_rank": prev_ranks,
|
||
|
|
"current_published_rank": cur_ranks
|
||
|
|
}).dropna()
|
||
|
|
|
||
|
|
comparison["rank_shift"] = comparison["previous_published_rank"] - comparison["current_published_rank"]
|
||
|
|
flagged = comparison[comparison["rank_shift"].abs() > threshold].sort_values(by="rank_shift", ascending=False)
|
||
|
|
|
||
|
|
print(f"[RegressionAuditor] Total positions analyzed: {len(comparison)} | Movements > {threshold}: {len(flagged)}")
|
||
|
|
return flagged
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
audit_ranking_regression()
|