"""p05_ideas.py: idea scoring, survivors, the dataset vet, and every external number used in the idea, novelty and journal sections, registered in the values file. Judgments (rubric weights, threshold, integer scores, validity scores, missed-work probabilities, confidence) are registered with kind "judgment" so the check script can list them as the only numbers that do not come from data or retrieved sources. Weighted totals are computed here. """ import json, re, sys, math, datetime, csv sys.path.insert(0, "/home/claude/mill/code") from vals import Values INP = "/home/claude/mill/inputs" OUT = "/home/claude/mill/outputs" V = Values(script="p05_ideas.py") ideas = json.load(open(f"{INP}/ideas.json")) S = json.load(open(f"{INP}/scores.json")) NOV = json.load(open(f"{INP}/novelty.json")) JR = json.load(open(f"{INP}/journals.json")) VET = json.load(open(f"{INP}/vet.json")) CRIT = ["novelty", "feasibility", "effect", "audience", "desk_risk", "cost"] J = {"population": "analyst judgment", "filter": "Phase 2", "statistic": "judgment"} # ---------------- rubric and scores ------------------------------------------------------------ w = S["rubric"]["weights"] assert abs(sum(w.values()) - 1) < 1e-9 and set(w) == set(CRIT) for k in CRIT: V.set(f"w_{k}", w[k], "weight", {**J, "filter": "rubric weights"}, kind="judgment") V.set("threshold", S["rubric"]["threshold"], "score", {**J, "filter": "survivor threshold on the weighted score"}, kind="judgment") V.set("score_min", 1, "score", {**J, "filter": "rubric scale"}, kind="judgment") V.set("score_max", 5, "score", {**J, "filter": "rubric scale"}, kind="judgment") V.set("n_ideas", len(ideas), "ideas", {"population": "inputs/ideas.json", "filter": "none", "statistic": "count"}) V.set("n_criteria", len(CRIT), "criteria", {"population": "rubric", "filter": "none", "statistic": "count"}) tot = {} for idea in ideas: i = idea["id"]; sc = S["scores"][i] for k in CRIT: assert isinstance(sc[k], int) and 1 <= sc[k] <= 5, (i, k) V.set(f"sc_{i.lower()}_{k}", sc[k], "score", {**J, "filter": f"{i}, {k}"}, kind="judgment") t = sum(w[k] * sc[k] for k in CRIT) tot[i] = t V.set(f"tot_{i.lower()}", round(t, 10), "score", {"population": "idea scores", "filter": i, "statistic": "weighted sum of the six criterion scores"}) NEAR = 0.15 # band below the threshold reported as near-survivor (judgment, registered below) surv = [i for i, t in tot.items() if t >= S["rubric"]["threshold"] - 1e-12] near = [i for i, t in tot.items() if S["rubric"]["threshold"] - NEAR - 1e-12 <= t < S["rubric"]["threshold"] - 1e-12] V.set("n_survivors", len(surv), "ideas", {"population": "idea scores", "filter": "weighted score at or above the threshold", "statistic": "count"}) V.set("n_near", len(near), "ideas", {"population": "idea scores", "filter": "weighted score within the near band below the threshold", "statistic": "count"}) V.set("near_band", NEAR, "score", {**J, "filter": "band below the threshold called near-survivor"}, kind="judgment") ranking = sorted(tot, key=lambda i: -tot[i]) json.dump({"totals": tot, "survivors": surv, "near": near, "ranking": ranking}, open(f"{OUT}/idea_scores.json", "w"), indent=1) for p, pk in S["packages"].items(): V.set(f"pkg_{p.lower()}_n", len(pk["ideas"]), "ideas", {"population": "packages", "filter": p, "statistic": "member ideas"}) V.set("n_packages", len(S["packages"]), "papers", {"population": "packages", "filter": "none", "statistic": "count"}) # ---------------- vet ------------------------------------------------------------------------------ for k, c in VET["components"].items(): V.set(f"vet_{k}", c["score"], "score", {**J, "filter": f"validity component {k}"}, kind="judgment") V.set("vet_mean", sum(c["score"] for c in VET["components"].values()) / len(VET["components"]), "score", {"population": "validity components", "filter": "all five", "statistic": "unweighted mean of the component judgments"}) # ---------------- novelty ------------------------------------------------------------------------ users = [u for u in NOV["dataset_users"] if u["id"] in ("U1", "U2", "U3")] V.set("n_users_published", sum(1 for u in users if "doi:" in u["citation"]), "users", {"population": "retrieved works using the dataset", "filter": "with a DOI", "statistic": "count"}) V.set("n_dataset_users", len(users), "users", {"population": "retrieved works and repositories using the dataset", "filter": "papers and code repositories, Kaggle notebooks excluded", "statistic": "count"}) V.set("n_users_repo", sum(1 for u in users if "doi:" not in u["citation"]), "users", {"population": "retrieved works and repositories using the dataset", "filter": "without a DOI (code repositories)", "statistic": "count"}) for p, rec in NOV["missed_work_probability"].items(): V.set(f"pmiss_{p.lower()}", rec["p"], "probability", {**J, "filter": f"probability that close prior work for {p} was missed"}, kind="judgment") V.set("conf_recommendation", S["confidence"]["recommendation"], "probability", {**J, "filter": "confidence in the recommendation"}, kind="judgment") V.set("conf_if_confirmed", S["confidence"]["if_confirmed"], "probability", {**J, "filter": "confidence after the stated checks come back clean"}, kind="judgment") V.set("conf_gate", S["confidence"]["gate"], "probability", {**J, "filter": "confidence that continuing past the data-quality gate is right"}, kind="judgment") V.set("min_runs_interpretable", 5, "runs", {**J, "filter": "fewest runs for which a run-bootstrap interval is read as an interval"}, kind="judgment") pub = datetime.datetime.strptime(V.get("doc_article_published"), "%d %B %Y").date() ret = datetime.date.fromisoformat(V.get("sources_retrieved")) V.set("days_since_pub", (ret - pub).days, "days", {"population": "external source dates", "filter": "S_article publication date to retrieval date", "statistic": "days"}) V.set("months_since_pub", math.ceil((ret - pub).days / 30.44), "months", {"population": "external source dates", "filter": "S_article publication date to retrieval date", "statistic": "months, rounded up"}) V.set("n_closest_s1", len(NOV["closest"]["S1"]), "works", {"population": "novelty.json", "filter": "S1", "statistic": "closest works listed"}) V.set("n_closest_s2", len(NOV["closest"]["S2"]), "works", {"population": "novelty.json", "filter": "S2", "statistic": "closest works listed"}) # query log tallies rows = list(csv.DictReader(open(f"{INP}/query_log.tsv"), delimiter="\t")) V.set("q_rows", len(rows), "entries", {"population": "inputs/query_log.tsv", "filter": "none", "statistic": "entries"}) oa = [r for r in rows if r["database"] == "OpenAlex" and r["batch"] in ("A", "B", "C")] V.set("q_openalex_answered", len(oa), "queries", {"population": "inputs/query_log.tsv", "filter": "OpenAlex batches A to C", "statistic": "queries answered"}) dbs = sorted({r["database"] for r in rows}) V.set("q_databases", len(dbs), "databases", {"population": "inputs/query_log.tsv", "filter": "none", "statistic": "distinct databases queried"}) refused = sum(int(m) for r in rows if r["batch"] == "D" for m in re.findall(r"429 for (\d+)", r["status"])) V.set("q_openalex_refused", refused, "queries", {"population": "inputs/query_log.tsv", "filter": "OpenAlex re-runs", "statistic": "queries refused with HTTP 429"}) # ---------------- journals: numbers parsed from verified page text ----------------------------- for j in JR["journals"]: jid = j["id"].lower() if j["tag"] == "UNVERIFIED": continue E = {"population": "external source", "filter": f"{j['name']} pages, retrieved {JR['retrieved']}", "statistic": "value as printed"} mt = j["metrics_text"] m = re.search(r"([\d.]+) CiteScore ([\d.]+) Impact Factor", mt) if m: V.set(f"j_{jid}_citescore", float(m.group(1)), "CiteScore", E, kind="external") V.set(f"j_{jid}_if", float(m.group(2)), "impact factor", E, kind="external") m = re.fullmatch(r"(\d{4}) Journal Citation Reports \(Clarivate Analytics, (\d{4})\)", j.get("if_release_quote", "")) if m: V.set(f"j_{jid}_if_year", int(m.group(1)), "year", E, kind="external") V.set(f"j_{jid}_jcr_pub_year", int(m.group(2)), "year", E, kind="external") m = re.search(r"Journal Impact Factor ([\d.]+) \((\d{4})\) 5-year Journal Impact Factor ([\d.]+)", mt) if m: V.set(f"j_{jid}_if", float(m.group(1)), "impact factor", E, kind="external") V.set(f"j_{jid}_if_year", int(m.group(2)), "year", E, kind="external") V.set(f"j_{jid}_if5", float(m.group(3)), "impact factor", E, kind="external") for key, txt in j["timeline_text"].items(): m = re.fullmatch(r"(\d+) days?", txt) if m: V.set(f"j_{jid}_{key}_days", int(m.group(1)), "days", E, kind="external") m = re.fullmatch(r"(\d+)%", txt) if m: V.set(f"j_{jid}_{key}", int(m.group(1)) / 100, "fraction", E, kind="external") for q in j["fee_quotes"]: m = re.search(r"USD (\d+)\b", q) or re.search(r"\$(\d+)\.00 USD", q) if m: V.set(f"j_{jid}_apc_usd", int(m.group(1)), "USD", E, kind="external") if j["id"] == "MFGLET": m = re.search(r"limited to (\d+) words.*maximum of four \((\d)\) figures", j["article_types"]) V.set("j_mfglet_word_limit", int(m.group(1)), "words", E, kind="external") V.set("j_mfglet_max_figs", int(m.group(2)), "figures or tables", E, kind="external") m = re.search(r"within four \((\d)\) weeks", j["review_rule"]) V.set("j_mfglet_decision_weeks", int(m.group(1)), "weeks", E, kind="external") if j["id"] == "JMP": m = re.search(r"From (\w+ \d+, \d{4}).*submitted after (\w+ \d+, \d{4})", j["fee_quotes"][0]) V.set("j_jmp_gold_from", m.group(1), "date", E, kind="external") V.set("j_jmp_apc_after", m.group(2), "date", E, kind="external") V.set("n_journals_verified", sum(1 for j in JR["journals"] if j["tag"] in ("VERIFIED", "PARTIAL")), "journals", {"population": "inputs/journals.json", "filter": "tag VERIFIED or PARTIAL", "statistic": "count"}) # ---------------- pre-specification record -------------------------------------------------------- ps = open(f"{OUT}/prespec_sha256.txt").read().split() V.set("prespec_sha256", ps[0], "sha256", {"population": "outputs/prespec_descriptive_questions.md", "filter": "file as frozen", "statistic": "sha256"}, kind="text") V.set("prespec_time", ps[2], "UTC time", {"population": "outputs/prespec_descriptive_questions.md", "filter": "file as frozen", "statistic": "time the hash was taken"}, kind="text") V.save() print("p05_ideas: survivors", surv, "near", near) print("ranking", [(i, round(tot[i], 2)) for i in ranking])