建立 DMTA 闭环：自动生成项目进展报告

DMTA（Design–Make–Test–Analyze）闭环的价值在于每一轮都基于上一轮的数据做更好的决策。但如果每轮的数据分析都靠人工拼 PPT，闭环就转不快。把项目进展报告自动化，是让 DMTA 真正转起来的工程基础。

报告应该回答什么

问题	对应指标
我们在进步吗？	各轮次的最佳/中位活性、性质达标率
哪个方向有效？	按骨架/取代基类型分组的活性趋势
瓶颈在哪？	各性质的达标情况，哪一项拖后腿
模型可信吗？	预测 vs 实测的回测结果
下一步测什么？	主动学习给出的候选
还差多远？	与 TPP 目标的差距

数据结构设计

import pandas as pd

# 项目数据的核心表结构
# compounds: 化合物注册
#   internal_id, smiles, inchikey, scaffold, round_designed, designer
#
# assays: 测定定义
#   assay_id, name, type, unit, direction (higher_better/lower_better),
#   target_value, protocol_version
#
# results: 实测数据
#   internal_id, assay_id, value, relation, batch, run_date, operator
#
# predictions: 模型预测（关键：与实测分开存！）
#   internal_id, assay_id, predicted_value, uncertainty,
#   model_version, predicted_date

# 「预测」与「实测」必须分表或有明确标记 ——
# 混在一起是危险的做法

自动化报告生成

#!/usr/bin/env python3
"""generate_report.py —— DMTA 轮次报告"""
import pandas as pd
import numpy as np
from datetime import datetime

def round_progress(results, compounds, assay="pIC50"):
    """各轮次的进展"""
    df = results[results["assay_id"] == assay].merge(
        compounds[["internal_id", "round_designed", "scaffold"]],
        on="internal_id")
    prog = df.groupby("round_designed")["value"].agg(
        n="size", best="max", median="median", mean="mean")
    prog["cumulative_best"] = prog["best"].cummax()
    return prog

def property_gate_analysis(results, gates):
    """哪些性质是瓶颈"""
    rows = []
    for assay, (target, direction) in gates.items():
        sub = results[results["assay_id"] == assay]
        if sub.empty:
            continue
        passed = ((sub["value"] >= target) if direction == "higher"
                  else (sub["value"] <= target))
        rows.append({"assay": assay, "n_tested": len(sub),
                     "n_passed": int(passed.sum()),
                     "pass_rate": float(passed.mean()),
                     "target": target})
    return pd.DataFrame(rows).sort_values("pass_rate")

def model_backtest(predictions, results, assay):
    """模型回测：预测 vs 后来的实测"""
    merged = predictions[predictions["assay_id"] == assay].merge(
        results[results["assay_id"] == assay][["internal_id", "value"]],
        on="internal_id", how="inner")
    if len(merged) < 5:
        return None
    err = merged["predicted_value"] - merged["value"]
    return {
        "n": len(merged),
        "rmse": float(np.sqrt((err ** 2).mean())),
        "mae": float(err.abs().mean()),
        "bias": float(err.mean()),
        "spearman": float(merged["predicted_value"].corr(
            merged["value"], method="spearman")),
        # 关键：不确定性是否可信
        "uncertainty_correlation": float(
            merged["uncertainty"].corr(err.abs())),
    }

def scaffold_analysis(results, compounds, assay="pIC50", min_n=3):
    """哪个骨架系列最有希望"""
    df = results[results["assay_id"] == assay].merge(
        compounds[["internal_id", "scaffold"]], on="internal_id")
    g = df.groupby("scaffold")["value"].agg(n="size", best="max",
                                            median="median")
    return g[g["n"] >= min_n].sort_values("best", ascending=False)

关键指标：模型回测

这是报告中最有价值也最常被省略的部分。每一轮实验结束后，把上一轮的预测与新拿到的实测数据对比：

bt = model_backtest(predictions, results, "pIC50")
print(f"回测样本数: {bt['n']}")
print(f"RMSE: {bt['rmse']:.3f}")
print(f"系统偏差: {bt['bias']:+.3f}")     # 明显非零说明模型有系统性高估/低估
print(f"排序相关: {bt['spearman']:.3f}")
print(f"不确定性有效性: {bt['uncertainty_correlation']:.3f}")

# 判读：
#   spearman > 0.5      → 模型对排序有帮助，可用于优先级
#   spearman ~ 0        → 模型没用，应停止依赖它
#   bias 明显非零        → 需要校准
#   uncertainty_corr > 0.3 → 不确定性估计有意义，可用于过滤
#   uncertainty_corr ~ 0   → 不确定性无意义，别用它做决策

这个回测是模型可信度的唯一真实来源。交叉验证的指标是在历史数据上算的，而回测是真正的前瞻验证。

生成 HTML 报告

import mols2grid
from jinja2 import Template

def build_html_report(data, out_path="dmta_report.html"):
    # 分子网格（可交互浏览）
    grid_html = mols2grid.display(
        data["top_compounds"], smiles_col="smiles",
        subset=["img", "internal_id", "pIC50", "logD", "HLM"],
        tooltip=["MW", "TPSA", "scaffold"],
        n_cols=5, size=(160, 130),
    )._repr_html_()

    template = Template("""
<h1>{{ project }} — 第 {{ round }} 轮报告</h1>
<p>生成时间：{{ date }}</p>

<h2>进展概览</h2>
{{ progress_table }}

<h2>性质瓶颈</h2>
{{ gate_table }}
<p>通过率最低的性质就是当前瓶颈，下一轮应重点改善。</p>

<h2>模型回测</h2>
{{ backtest_table }}

<h2>骨架系列表现</h2>
{{ scaffold_table }}

<h2>本轮最佳化合物</h2>
{{ grid }}

<h2>下一轮建议</h2>
<ul>{% for s in suggestions %}<li>{{ s }}</li>{% endfor %}</ul>
""")
    html = template.render(
        project=data["project"], round=data["round"],
        date=datetime.now().strftime("%Y-%m-%d"),
        progress_table=data["progress"].to_html(),
        gate_table=data["gates"].to_html(),
        backtest_table=pd.DataFrame([data["backtest"]]).to_html(),
        scaffold_table=data["scaffolds"].to_html(),
        grid=grid_html,
        suggestions=data["suggestions"],
    )
    open(out_path, "w").write(html)
    return out_path

下一轮建议：主动学习

def suggest_next_batch(candidates, model_ensemble, X_cand,
                       target=8.0, batch_size=24):
    """结合预测值与不确定性挑选下一批"""
    preds = np.stack([m.predict(X_cand) for m in model_ensemble])
    mean, std = preds.mean(axis=0), preds.std(axis=0)

    # 采集函数：既要可能好（exploitation），
    #          也要信息量大（exploration）
    ucb = mean + 1.0 * std                       # 上置信界

    picked = []
    # 一半选 UCB 最高的（推进项目）
    picked += list(np.argsort(ucb)[-batch_size // 2:])
    # 一半选不确定性最高且预测尚可的（改善模型）
    mask = mean > np.percentile(mean, 50)
    explore = np.where(mask)[0]
    explore = explore[np.argsort(std[explore])[-batch_size // 2:]]
    picked += list(explore)

    return sorted(set(picked))

# 再用多样性采样确保覆盖不同化学型（见 331）

延伸资源

• 工程化：225《AI 制药工具工程化》；主动学习：283《Optibrium Cerella》；
• 可视化：217《Mols2grid》；MPO 与决策见「决策与监管」模块。