#!/usr/bin/env python3 """DMCausalTriangulate-Kor — CLI entry point. Works without streamlit/plotly/pandas installed. Uses stdlib + the triangulation package only. Examples: python3 main.py --help python3 main.py --grid python3 main.py --pair "SGLT2i" "CV death" python3 main.py --discordant --top 5 python3 main.py --report out.json python3 main.py --pair "DPP-4i" "pancreatitis" --report-md out.md 연구·교육용 도구. 임상의사결정에 직접 사용 금지. """ from __future__ import annotations import argparse import json import os import sys from typing import Any # Allow running this file from anywhere HERE = os.path.dirname(os.path.abspath(__file__)) if HERE not in sys.path: sys.path.insert(0, HERE) DISCLAIMER = ( "본 도구는 연구 가설 생성·문헌 갭 분석 목적의 연구용·교육용 도구입니다. " "임상의사결정·환자 진료에 직접 사용해서는 안 됩니다." ) def _print_grid(rows: list[dict[str, Any]]) -> None: print(f"{'drug_class':<14} {'outcome':<22} {'designs':>7} {'conc':>6} {'maj':>4} {'agr':>5} {'#eff':>5}") print("-" * 70) for r in rows: print( f"{r['drug_class']:<14} {r['outcome']:<22} " f"{r['designs_present']:>7} {r['concordance']:>6} " f"{r['majority_direction']:>4} {r['direction_agreement']:>5} " f"{r['n_effects']:>5}" ) def _print_pair(drug_class: str, outcome: str, data_dir: str | None) -> dict[str, Any]: from triangulation.ontology import load_effects, filter_pair from triangulation.grid import concordance_score, build_grid from triangulation.lawlor import score_lawlor_criteria from triangulation.bias import diagnose_discordance from triangulation.designs import recommend_designs effects = load_effects(data_dir) if data_dir else load_effects() pair_effects = filter_pair(effects, drug_class, outcome) if not pair_effects: print(f"No effects found for pair ({drug_class!r}, {outcome!r}).", file=sys.stderr) return {} conc = concordance_score(pair_effects) grid = build_grid(pair_effects) lawlor = score_lawlor_criteria(pair_effects) biases = diagnose_discordance(pair_effects) designs = recommend_designs(biases, drug_class, outcome) print(f"=== Pair: {drug_class} × {outcome} ===") print() print(f"Effects: {len(pair_effects)} rows across {len(grid)} designs ({', '.join(sorted(grid))})") print(f"Concordance: {conc['concordance']} (direction-agreement={conc['direction_agreement']}, " f"CI-overlap={conc['ci_overlap_fraction']}, designs_present={conc['designs_present']}, " f"majority={conc['majority_direction']})") print() print(f"Lawlor 2016 5-criterion: total {lawlor['total']}/{lawlor['max']} " f"(normalized {lawlor['normalized']})") for c in lawlor["criteria"]: print(f" - {c['name']}: {c['score']}/{c['max']} — {c['rationale']}") print() if biases: print("Discordance bias diagnosis (ranked):") for b in biases: print(f" - [{b['score']}] {b['bias_type']}: {b['rationale']}") else: print("No discordance pattern triggered (designs are concordant).") print() print(f"Follow-up design cards (top {len(designs)}):") for i, c in enumerate(designs, 1): print(f" {i}. {c['name']} (match {c['match_score']})") print(f" endpoint: {c['primary_endpoint']}") print(f" n≈{c['estimated_sample_size']} follow-up≈{c['follow_up_years']}y") print(f" hypothesis: {c['key_hypothesis']}") print() print(f"NOTE: {DISCLAIMER}") return { "pair": [drug_class, outcome], "effects": pair_effects, "concordance": conc, "lawlor": lawlor, "biases": biases, "designs": designs, } def _discordant_summary(top: int, data_dir: str | None) -> list[dict[str, Any]]: from triangulation.ontology import load_effects, filter_pair from triangulation.grid import grid_summary from triangulation.bias import diagnose_discordance effects = load_effects(data_dir) if data_dir else load_effects() rows = grid_summary(effects) enriched: list[dict[str, Any]] = [] for r in rows: pair = filter_pair(effects, r["drug_class"], r["outcome"]) biases = diagnose_discordance(pair) r2 = dict(r) r2["n_bias_hits"] = len(biases) r2["top_bias"] = biases[0]["bias_type"] if biases else None r2["top_bias_score"] = biases[0]["score"] if biases else 0 enriched.append(r2) # Sort: lowest concordance + bias hits first enriched.sort(key=lambda r: (r["concordance"], -r["top_bias_score"], -r["n_bias_hits"])) return enriched[:top] def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser( prog="dm-causal-triangulate", description=( "DMCausalTriangulate-Kor — 5-design causal triangulation for T2DM " "drug-outcome grids. " + DISCLAIMER ), ) p.add_argument("--data-dir", default=None, help="Override data directory") p.add_argument("--grid", action="store_true", help="Print summary grid of all pairs") p.add_argument( "--pair", nargs=2, metavar=("DRUG_CLASS", "OUTCOME"), help='Analyze one pair, e.g. --pair "SGLT2i" "CV death"', ) p.add_argument("--discordant", action="store_true", help="List most discordant pairs") p.add_argument("--top", type=int, default=5, help="Top-N for --discordant") p.add_argument( "--report", metavar="PATH", help="When used with --pair, write a JSON report. With --grid, write the grid summary.", ) p.add_argument( "--report-md", metavar="PATH", help="When used with --pair, write a Markdown report.", ) p.add_argument( "--report-docx", metavar="PATH", help="When used with --pair, write a docx (falls back to .md without python-docx).", ) args = p.parse_args(argv) if not any([args.grid, args.pair, args.discordant]): p.print_help() return 0 if args.grid: from triangulation.ontology import load_effects from triangulation.grid import grid_summary effects = load_effects(args.data_dir) if args.data_dir else load_effects() rows = grid_summary(effects) rows.sort(key=lambda r: (-r["concordance"], r["drug_class"], r["outcome"])) _print_grid(rows) if args.report: with open(args.report, "w", encoding="utf-8") as f: json.dump(rows, f, indent=2, ensure_ascii=False) print(f"\nWrote {len(rows)} rows to {args.report}") return 0 if args.discordant: rows = _discordant_summary(args.top, args.data_dir) print(f"Top {len(rows)} most discordant pairs:") for r in rows: print( f" {r['drug_class']:<12} × {r['outcome']:<22} " f"conc={r['concordance']:<5} top_bias={r['top_bias'] or '-':<28} " f"({r['top_bias_score']})" ) if args.report: with open(args.report, "w", encoding="utf-8") as f: json.dump(rows, f, indent=2, ensure_ascii=False) print(f"\nWrote {len(rows)} rows to {args.report}") return 0 if args.pair: drug_class, outcome = args.pair result = _print_pair(drug_class, outcome, args.data_dir) if not result: return 1 if args.report: with open(args.report, "w", encoding="utf-8") as f: json.dump(result, f, indent=2, ensure_ascii=False) print(f"\nWrote JSON report to {args.report}") if args.report_md or args.report_docx: from triangulation.report import build_markdown_report, build_docx_report if args.report_md: md = build_markdown_report( (drug_class, outcome), result["effects"], result["concordance"], result["lawlor"], result["biases"], result["designs"], ) with open(args.report_md, "w", encoding="utf-8") as f: f.write(md) print(f"Wrote Markdown report to {args.report_md}") if args.report_docx: actual = build_docx_report( (drug_class, outcome), result["effects"], result["concordance"], result["lawlor"], result["biases"], result["designs"], args.report_docx, ) print(f"Wrote report to {actual}") return 0 return 0 if __name__ == "__main__": sys.exit(main())