P3-5.7 折叠多个后续事件的规则¶

# 这个例子把同一样本之后的多个后续事件折叠进表结构，并确定代表标签。
import csv
from collections import defaultdict
from pathlib import Path

sample_roster_path = Path("docs/assets/part-03/chapter-05/p3_5_7_sample_roster.csv")
follow_up_events_path = Path("docs/assets/part-03/chapter-05/p3_5_7_follow_up_events.csv")
event_severity_path = Path("docs/assets/part-03/chapter-05/p3_5_7_event_severity.csv")

selected_severity_cutoff = 4
severity_cutoffs = [4, 3, 2]
follow_up_window_days = 7
failure_types = {"failure", "critical_failure"}
preview_row_count = 12

def read_csv(path):
    with path.open(newline="", encoding="utf-8") as file:
        return list(csv.DictReader(file))

sample_roster = read_csv(sample_roster_path)
follow_ups = read_csv(follow_up_events_path)
severity_table = read_csv(event_severity_path)
severity_by_type = {row["event_type"]: int(row["severity"]) for row in severity_table}

for row in follow_ups:
    row["days_after_sample"] = int(row["days_after_sample"])
    row["severity"] = severity_by_type[row["event_type"]]

period_events = [row for row in follow_ups if 1 <= row["days_after_sample"] <= follow_up_window_days]
ordered_events = sorted(period_events, key=lambda row: (row["sample_id"], row["days_after_sample"]))
events_by_sample = defaultdict(list)
for row in ordered_events:
    events_by_sample[row["sample_id"]].append(row)

folded = []
for sample in sample_roster:
    sample_id = sample["sample_id"]
    events = events_by_sample.get(sample_id, [])
    if events:
        first_event = events[0]["event_type"]
        worst = sorted(events, key=lambda row: (-row["severity"], row["days_after_sample"]))[0]
        worst_event = worst["event_type"]
        worst_severity = worst["severity"]
        event_sequence = " > ".join(row["event_type"] for row in events)
    else:
        first_event = "none"
        worst_event = "none"
        worst_severity = 0
        event_sequence = "none"
    folded.append(
        {
            "sample_id": sample_id,
            "first_event": first_event,
            "worst_event": worst_event,
            "worst_severity": worst_severity,
            "event_count": len(events),
            "event_sequence": event_sequence,
            "any_failure": int(any(row["event_type"] in failure_types for row in events)),
            "any_selected_event": int(any(row["severity"] >= selected_severity_cutoff for row in events)),
        }
    )

cutoff_results = []
for cutoff in severity_cutoffs:
    selected = [row for row in folded if row["event_count"] > 0 and row["worst_severity"] >= cutoff]
    cutoff_results.append(
        {
            "severity_cutoff": cutoff,
            "selected_sample_count": len(selected),
            "selected_samples": ",".join(row["sample_id"] for row in selected) or "none",
        }
    )

print("1) raw follow-up events")
print("sample_id  days_after_sample       event_type source_system")
for row in follow_ups[:preview_row_count]:
    print(
        f"{row['sample_id']:>9} {row['days_after_sample']:>18} "
        f"{row['event_type']:>16} {row['source_system']:>13}"
    )
print(f"... {len(follow_ups) - preview_row_count} more follow-up events")
print()
print("2) severity rule table")
print("      event_type  severity")
for row in severity_table[:preview_row_count]:
    print(f"{row['event_type']:>16} {int(row['severity']):>9}")
print(f"... {len(severity_table) - preview_row_count} more severity rules")
print()
print(f"3) folded result when severity_cutoff = {selected_severity_cutoff}")
print(
    "sample_id      first_event      worst_event  worst_severity  event_count"
    "             event_sequence  any_failure  any_selected_event"
)
for row in folded[:preview_row_count]:
    print(
        f"{row['sample_id']:>9} {row['first_event']:>16} {row['worst_event']:>16} "
        f"{row['worst_severity']:>15} {row['event_count']:>12} "
        f"{row['event_sequence']:>26} {row['any_failure']:>12} {row['any_selected_event']:>19}"
    )
print(f"... {len(folded) - preview_row_count} more folded samples")
print()
print("4) sensitivity by severity_cutoff")
print(
    " severity_cutoff  selected_sample_count"
    "                                                                     selected_samples"
)
for row in cutoff_results:
    print(
        f"{row['severity_cutoff']:>24} {row['selected_sample_count']:>21} "
        f"{row['selected_samples']:>83}"
    )