run_my_data.py

# -*- coding: utf-8 -*-
"""run_my_data.py — 用自己的数据运行统一信息论小样本工具（模板）

使用步骤：
  1) 把数据文件（CSV）放到 unified_loop/ 目录下，或修改下面 DATA 路径；
  2) 填写 COMP_COLS（组分列名）和 TARGET（目标列名）；
  3) 运行：python run_my_data.py
  4) 输出：results/my_retrospective.csv（回测：能省多少实验）
          results/my_recommendations.csv（下一批推荐 Top-N）
          results/my_recommendation_report.md（人话版报告）

数据要求：
  - 每行 = 一个配方/实验；组分列 = 各原料用量（没放的组分留空或填 0）；
  - 目标列 = 连续数值性能（越大越好，工具自动做最大化；要最小化请改 maximize=False）；
  - 至少 20~50 条有目标值的历史数据。
"""
import time
from pathlib import Path

import pandas as pd

from smallmatprep.active.loop import (
    generate_candidate_formulations,
    recommend_next_experiments,
    run_retrospective_loop,
)
from smallmatprep.active.report import write_recommendation_report

RESULTS = Path(__file__).parent / "results"

# ============ 这里改成你自己的配置 ============
DATA = Path(__file__).parent / "your_data.csv"   # ← 你的数据文件路径
COMP_COLS = None                                 # ← 组分列名列表，如 ["A","B","C"]；None = 除目标列外全是组分
TARGET = "P2"                                    # ← 你的目标列名
N_INITIAL, BUDGET, N_SEED = 8, 12, 12            # 回测配置：初始实验数 / 预算 / 随机种子数
N_CANDIDATES, N_RECOMMEND = 200, 10              # 生成候选数 / 推荐条数
# ==============================================


def main():
    t0 = time.time()
    RESULTS.mkdir(parents=True, exist_ok=True)
    df = pd.read_csv(DATA, encoding="utf-8-sig")
    df = df.dropna(subset=[TARGET]).reset_index(drop=True)
    comp_cols = list(COMP_COLS) if COMP_COLS else [c for c in df.columns if c != TARGET]
    X = df[comp_cols].fillna(0.0).values.astype(float)   # 结构零（没放的组分）→ 0
    y = df[TARGET].values.astype(float)
    print(f"数据: {len(y)} 个配方 x {X.shape[1]} 个组分 | {TARGET} 范围 [{y.min():.2f}, {y.max():.2f}]")

    # 1) 回溯闭环：这个工具在你的数据上能省多少实验（auto vs random）
    print("\n=== 回溯闭环（auto vs random）===")
    rows = []
    for strat in ["auto", "random"]:
        for seed in range(N_SEED):
            r = run_retrospective_loop(X, y, n_initial=N_INITIAL, budget=BUDGET,
                                       strategy=strat, seed=seed, target_ratio=0.95)
            rows.append({"strategy": r.strategy, "seed": seed, "best": r.best_value,
                         "success95": r.success, "steps": r.steps_to_target})
    retro = pd.DataFrame(rows)
    retro.to_csv(RESULTS / "my_retrospective.csv", index=False, encoding="utf-8-sig")
    for strat in ["auto", "random"]:
        sub = retro[retro["strategy"] == strat]
        st = sub["steps"].dropna()
        print(f"  {strat:8s} mean_best={sub['best'].mean():.2f} "
              f"succ95={sub['success95'].mean() * 100:.0f}% "
              f"中位步数={st.median() if len(st) else float('nan'):.1f}")

    # 2) 生成候选配方 + 推荐下一批实验
    cand = generate_candidate_formulations(X, n_candidates=N_CANDIDATES, seed=2026)
    rec = recommend_next_experiments(X, y, cand, n_recommend=N_RECOMMEND,
                                     strategy="auto", seed=0, feature_names=comp_cols)
    rec.to_csv(RESULTS / "my_recommendations.csv", index=False, encoding="utf-8-sig")
    write_recommendation_report(rec, target=TARGET, strategy=rec.attrs.get("strategy", "auto"),
                                path=str(RESULTS / "my_recommendation_report.md"), n_history=len(X))
    print("\n=== 下一批实验推荐 Top-5 ===")
    print(rec.head(5)[["rank", "score", "pred_mean", "pred_std", "formulation"]].to_string(index=False))
    print(f"\n报告: {RESULTS / 'my_recommendation_report.md'}")
    print(f"总耗时 {time.time() - t0:.1f}s")


if __name__ == "__main__":
    main()