#!/usr/bin/env python3 """Materialize a deterministic public-equity-investing idea scorecard from supplied CSV inputs.""" from __future__ import annotations import argparse import csv import json import math from datetime import date, datetime from pathlib import Path from typing import Any SCORE_WEIGHTS = { "variant_perception_score": 1.25, "catalyst_score": 1.15, "valuation_score": 1.10, "revisions_score": 1.00, "quality_score": 0.90, "growth_score": 0.85, "risk_reward_score": 1.15, "liquidity_score": 0.55, "crowding_score": 0.55, "momentum_score": 0.50, "portfolio_fit_score": 0.70, } OUTPUT_FIELDS = [ "rank", "ticker", "security", "company", "idea_type", "direction", "sector", "composite_score", "bucket", "valid_score_count", "variant_view", "catalyst", "first_rejection_risk", "next_step", "source", "source_as_of", "freshness_status", "warnings", ] def parse_score(value: Any, field: str, warnings: list[str]) -> float | None: text = str(value or "").strip().replace(",", "") if not text: return None try: score = float(text) except ValueError: warnings.append(f"{field} invalid numeric value {value!r}") return None if not math.isfinite(score): warnings.append(f"{field} is not finite") return None if score < 0: warnings.append(f"{field} negative score excluded") return None if score <= 5: return score / 5.0 * 100.0 if score <= 100: return score warnings.append(f"{field} above 100 capped at 100") return 100.0 def bucket_for(score: float | None) -> str: if score is None: return "needs_scoring_inputs" if score >= 75: return "A - immediate research candidate" if score >= 60: return "B - watchlist / needs trigger" if score >= 45: return "C - screen flag only" return "Reject / low-priority false positive" def parse_iso_date(value: Any, field: str, warnings: list[str]) -> date | None: text = str(value or "").strip() if not text: return None try: return datetime.strptime(text, "%Y-%m-%d").date() except ValueError: warnings.append(f"{field} invalid date {value!r}; expected YYYY-MM-DD") return None def source_freshness(row: dict[str, str], run_date: date, warnings: list[str]) -> tuple[str, str]: source_as_of = str( row.get("source_as_of") or row.get("source_date") or row.get("as_of_date") or "" ).strip() if not source_as_of: warnings.append("missing source_as_of") return "", "missing" parsed = parse_iso_date(source_as_of, "source_as_of", warnings) if parsed is None: return source_as_of, "invalid" if parsed > run_date: warnings.append("source_as_of is after run_date") return source_as_of, "future" if (run_date - parsed).days > 180: warnings.append("source_as_of is stale") return source_as_of, "stale" return source_as_of, "current" def load_rows(path: Path) -> list[dict[str, str]]: with path.open(newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) if not reader.fieldnames: raise ValueError("input CSV must include a header row") rows = [dict(row) for row in reader] if not rows: raise ValueError("input CSV must include at least one candidate row") return rows def score_row(row: dict[str, str], index: int, run_date: date) -> dict[str, Any]: warnings: list[str] = [] ticker = str(row.get("ticker") or "").strip() security = str(row.get("security") or "").strip() if not ticker and not security: raise ValueError(f"row {index}: ticker or security is required") weighted_score = 0.0 weight_total = 0.0 valid_count = 0 for field, weight in SCORE_WEIGHTS.items(): score = parse_score(row.get(field), field, warnings) if score is None: continue weighted_score += score * weight weight_total += weight valid_count += 1 composite = round(weighted_score / weight_total, 2) if weight_total else None if valid_count == 0: warnings.append("no valid numeric score fields provided") if not str(row.get("variant_view") or "").strip(): warnings.append("missing variant_view") if not str(row.get("first_rejection_risk") or "").strip(): warnings.append("missing first_rejection_risk") source_as_of, freshness_status = source_freshness(row, run_date, warnings) return { "rank": 0, "ticker": ticker, "security": security, "company": str(row.get("company") or "").strip(), "idea_type": str(row.get("idea_type") or "").strip(), "direction": str(row.get("direction") or "").strip(), "sector": str(row.get("sector") or "").strip(), "composite_score": composite, "bucket": bucket_for(composite), "valid_score_count": valid_count, "variant_view": str(row.get("variant_view") or "").strip(), "catalyst": str(row.get("catalyst") or "").strip(), "first_rejection_risk": str(row.get("first_rejection_risk") or "").strip(), "next_step": str(row.get("next_step") or "").strip(), "source": str(row.get("source") or "").strip(), "source_as_of": source_as_of, "freshness_status": freshness_status, "warnings": "; ".join(warnings), } def materialize(rows: list[dict[str, str]], run_date: date | None = None) -> list[dict[str, Any]]: run_date = run_date or date.today() scored = [score_row(row, idx, run_date) for idx, row in enumerate(rows, start=2)] scored.sort( key=lambda row: ( -1 if row["composite_score"] is None else -row["composite_score"], row["ticker"] or row["security"], ) ) for rank, row in enumerate(scored, start=1): row["rank"] = rank return scored def write_csv(path: Path, rows: list[dict[str, Any]]) -> None: with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=OUTPUT_FIELDS) writer.writeheader() for row in rows: writer.writerow( { field: "" if row.get(field) is None else row.get(field, "") for field in OUTPUT_FIELDS } ) def markdown(rows: list[dict[str, Any]]) -> str: lines = [ "# Public Equity Investing Idea Scorecard", "", "> Deterministic materializer from supplied inputs only. Scores rank research candidates, not final recommendations.", "", "| Rank | Ticker/Security | Type | Score | Bucket | Variant View | Catalyst | First Rejection Risk | Warnings |", "|---:|---|---|---:|---|---|---|---|---|", ] for row in rows: label = row["ticker"] or row["security"] score = "" if row["composite_score"] is None else f"{row['composite_score']:.2f}" lines.append( f"| {row['rank']} | {label} | {row['idea_type']} | {score} | {row['bucket']} | {row['variant_view']} | {row['catalyst']} | {row['first_rejection_risk']} | {row['warnings']} |" ) return "\n".join(lines) + "\n" def main() -> int: parser = argparse.ArgumentParser( description="Score supplied public-equity idea candidates into a deterministic PM triage log." ) parser.add_argument("input_csv", type=Path) parser.add_argument( "--output-dir", type=Path, default=Path("/tmp/public_equity_investing_idea_generation_output"), ) parser.add_argument( "--run-date", default=date.today().isoformat(), help="Freshness date for source_as_of checks, YYYY-MM-DD.", ) args = parser.parse_args() try: run_date = datetime.strptime(args.run_date, "%Y-%m-%d").date() rows = materialize(load_rows(args.input_csv), run_date) except Exception as exc: print(f"ERROR: {exc}") return 1 args.output_dir.mkdir(parents=True, exist_ok=True) write_csv(args.output_dir / "ranked_ideas.csv", rows) (args.output_dir / "idea_scorecard.json").write_text( json.dumps({"rows": rows}, indent=2) + "\n", encoding="utf-8" ) (args.output_dir / "idea_scorecard_support_note.md").write_text( markdown(rows), encoding="utf-8" ) print(f"Wrote idea scorecard outputs to {args.output_dir}") return 0 if __name__ == "__main__": raise SystemExit(main())