P3-4.5 现在收集到的样本，在多大程度上代表了整体运行情况¶

# 这个例子按类别和时间区段检查已收集样本对整体运行情况的代表性。
import csv
from collections import Counter
from pathlib import Path

minimum_count = 9
preview_sample_count = 8

input_path = Path("docs/assets/part-03/chapter-04/p3_4_5_sample_coverage.csv")
coverage_scopes = ["shift", "load_mode", "machine_id", "maintenance_phase"]

with input_path.open(newline="", encoding="utf-8") as file:
    samples = list(csv.DictReader(file))

coverage_summary = []
for scope in coverage_scopes:
    counts = Counter(sample[scope] for sample in samples)
    ordered_counts = sorted(counts.items(), key=lambda item: (-item[1], item[0]))
    most_seen, most_seen_count = ordered_counts[0]
    least_seen, _ = sorted(counts.items(), key=lambda item: (item[1], item[0]))[0]
    under_minimum = sum(1 for count in counts.values() if count < minimum_count)
    coverage_summary.append(
        {
            "scope": scope,
            "most_seen": most_seen,
            "count": most_seen_count,
            "least_seen": least_seen,
            "unique_conditions": len(counts),
            "under_minimum_conditions": under_minimum,
        }
    )

print("1) raw sample coverage table")
for sample in samples[:preview_sample_count]:
    print(
        f"{sample['event_id']}: shift={sample['shift']}, "
        f"load_mode={sample['load_mode']}, machine_id={sample['machine_id']}, "
        f"maintenance_phase={sample['maintenance_phase']}"
    )
print(f"... {len(samples) - preview_sample_count} more event-level samples")
print()
print(f"2) coverage summary when minimum_count = {minimum_count}")
for item in coverage_summary:
    print(
        f"{item['scope']}: most_seen={item['most_seen']} ({item['count']}), "
        f"least_seen={item['least_seen']}, "
        f"unique_conditions={item['unique_conditions']}, "
        f"under_minimum_conditions={item['under_minimum_conditions']}"
    )

1) raw sample coverage table
E01: shift=day, load_mode=normal, machine_id=M1, maintenance_phase=stable
E02: shift=day, load_mode=normal, machine_id=M1, maintenance_phase=stable
E03: shift=day, load_mode=normal, machine_id=M1, maintenance_phase=stable
E04: shift=day, load_mode=normal, machine_id=M1, maintenance_phase=stable
E05: shift=day, load_mode=normal, machine_id=M1, maintenance_phase=stable
E06: shift=day, load_mode=normal, machine_id=M1, maintenance_phase=stable
E07: shift=day, load_mode=normal, machine_id=M1, maintenance_phase=stable
E08: shift=day, load_mode=normal, machine_id=M1, maintenance_phase=stable
... 28 more event-level samples

2) coverage summary when minimum_count = 9
shift: most_seen=day (26), least_seen=night, unique_conditions=2, under_minimum_conditions=0
load_mode: most_seen=normal (25), least_seen=low, unique_conditions=3, under_minimum_conditions=2
machine_id: most_seen=M1 (22), least_seen=M2, unique_conditions=3, under_minimum_conditions=2
maintenance_phase: most_seen=stable (28), least_seen=after-maintenance, unique_conditions=2, under_minimum_conditions=1

# 这个例子用来确认训练覆盖偏斜时，基准模型和模型错误会集中在哪里。
import pandas as pd
from pathlib import Path
from sklearn.compose import ColumnTransformer
from sklearn.dummy import DummyClassifier
from sklearn.metrics import accuracy_score
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import OneHotEncoder
from sklearn.tree import DecisionTreeClassifier

input_path = Path("docs/assets/part-03/chapter-04/p3_4_5_sample_coverage.csv")
samples = pd.read_csv(input_path)

# 这是本节观察用的简化标签。真实运行标签需要另外经过业务检视来定义。
samples["needs_review"] = (
    samples["load_mode"].eq("high") | samples["maintenance_phase"].eq("after-maintenance")
).astype(int)

train = samples[samples["event_id"].between("E01", "E24")]
test = samples[samples["event_id"].between("E25", "E36")]
features = ["shift", "load_mode", "machine_id", "maintenance_phase"]

preprocess = ColumnTransformer(
    [("category", OneHotEncoder(handle_unknown="ignore"), features)]
)
models = {
    "dummy": DummyClassifier(strategy="most_frequent"),
    "tree": DecisionTreeClassifier(random_state=0, max_depth=3),
}

print("train coverage")
print(train.groupby(["load_mode", "maintenance_phase"])["event_id"].count())
print()
print("test coverage")
print(test.groupby(["load_mode", "maintenance_phase"])["event_id"].count())
print()

for name, estimator in models.items():
    model = make_pipeline(preprocess, estimator)
    model.fit(train[features], train["needs_review"])
    predicted = model.predict(test[features])
    result = test.assign(
        predicted=predicted,
        error=lambda df: df["predicted"].ne(df["needs_review"]),
    )
    print(f"{name} accuracy:", accuracy_score(test["needs_review"], predicted))
    print("errors by load_mode:", result.groupby("load_mode")["error"].sum().to_dict())

train coverage
load_mode  maintenance_phase
high       stable                4
normal     after-maintenance     2
           stable               18
Name: event_id, dtype: int64

test coverage
load_mode  maintenance_phase
high       after-maintenance    1
           stable               1
low        after-maintenance    2
           stable               3
normal     after-maintenance    3
           stable               2
Name: event_id, dtype: int64

dummy accuracy: 0.4166666666666667
errors by load_mode: {'high': 2, 'low': 2, 'normal': 3}
tree accuracy: 0.75
errors by load_mode: {'high': 0, 'low': 3, 'normal': 0}