371 lines
15 KiB
Python
371 lines
15 KiB
Python
from __future__ import annotations
|
|
import statistics
|
|
from collections import defaultdict
|
|
|
|
from app.models.action import ActionEntry, ActionType
|
|
from app.models.agent import AgentPersona
|
|
from app.models.world import WorldMetrics
|
|
|
|
|
|
METRIC_LABELS = {
|
|
"stability": "Stability",
|
|
"prosperity": "Prosperity",
|
|
"trust": "Trust",
|
|
"freedom": "Freedom",
|
|
"conflict": "Conflict",
|
|
"brand_sentiment": "Brand Sentiment",
|
|
"purchase_intent": "Purchase Intent",
|
|
"word_of_mouth": "Word of Mouth",
|
|
"churn_risk": "Churn Risk",
|
|
"adoption_rate": "Adoption Rate",
|
|
"information_spread": "Information Spread",
|
|
"echo_chamber_index": "Echo Chamber Index",
|
|
"rumor_distortion": "Rumor Distortion",
|
|
}
|
|
|
|
INVERSE_METRICS = {"conflict", "churn_risk"}
|
|
|
|
MARKET_METRICS = {"brand_sentiment", "purchase_intent", "word_of_mouth", "churn_risk", "adoption_rate"}
|
|
INFORMATION_METRICS = {"information_spread", "echo_chamber_index", "rumor_distortion"}
|
|
|
|
SOCIAL_METRICS = {"stability", "prosperity", "trust", "freedom", "conflict"}
|
|
|
|
|
|
def _rate(value_100: float, inverse: bool) -> str:
|
|
v = (100 - value_100) if inverse else value_100
|
|
if v >= 81:
|
|
return "excellent"
|
|
if v >= 61:
|
|
return "strong"
|
|
if v >= 41:
|
|
return "moderate"
|
|
if v >= 21:
|
|
return "weak"
|
|
return "critical"
|
|
|
|
|
|
def _trend(start: float, end: float) -> str:
|
|
delta = end - start
|
|
if delta > 0.05:
|
|
return "up"
|
|
if delta < -0.05:
|
|
return "down"
|
|
return "flat"
|
|
|
|
|
|
class ReportAnalyzer:
|
|
|
|
@staticmethod
|
|
def compute_metric_stats(metrics_history: list[dict], is_market: bool) -> list[dict]:
|
|
if not metrics_history:
|
|
return []
|
|
|
|
# Metrics are persisted as a complete WorldMetrics snapshot. Keep the
|
|
# report scorecard complete as well; hiding market or information
|
|
# fields based on the scenario made readers mistake omitted fields for
|
|
# missing data.
|
|
active_keys = set(METRIC_LABELS)
|
|
|
|
results = []
|
|
for key in active_keys:
|
|
values = [m.get(key, 0.5) for m in metrics_history]
|
|
start_val = values[0]
|
|
end_val = values[-1]
|
|
volatility = statistics.stdev(values) if len(values) > 1 else 0.0
|
|
|
|
value_100 = round(end_val * 100)
|
|
start_100 = round(start_val * 100)
|
|
inverse = key in INVERSE_METRICS
|
|
|
|
results.append({
|
|
"metric": key,
|
|
"label": METRIC_LABELS[key],
|
|
"value": value_100,
|
|
"start_value": start_100,
|
|
"end_raw": end_val,
|
|
"start_raw": start_val,
|
|
"min": round(min(values) * 100),
|
|
"max": round(max(values) * 100),
|
|
"trend": _trend(start_val, end_val),
|
|
"volatility": round(volatility, 4),
|
|
"rating": _rate(value_100, inverse),
|
|
"inverse": inverse,
|
|
})
|
|
|
|
return sorted(results, key=lambda x: x["metric"])
|
|
|
|
@staticmethod
|
|
def detect_data_gaps(
|
|
metrics_history: list[dict],
|
|
actions: list[ActionEntry],
|
|
is_market: bool,
|
|
tension_event_count: int = 0,
|
|
fallback_count: int = 0,
|
|
) -> list[dict]:
|
|
"""Find deterministic limitations worth surfacing in a report.
|
|
|
|
These are observations about this run, not claims that the simulator
|
|
is universally incapable of producing the signal. LLM prose can then
|
|
explain the finding without inventing a missing data source.
|
|
"""
|
|
gaps: list[dict] = []
|
|
if not metrics_history:
|
|
return [{
|
|
"key": "metrics_history",
|
|
"gap": "No metric history was recorded.",
|
|
"why_it_matters": "The report cannot establish trends or inflection points.",
|
|
"suggested_fix": "Check persistence and rerun the simulation.",
|
|
}]
|
|
|
|
action_types = {a.action_type.value for a in actions}
|
|
values_by_key = {
|
|
key: [float(row.get(key, 0.5)) for row in metrics_history]
|
|
for key in METRIC_LABELS
|
|
}
|
|
|
|
def add(key: str, gap: str, why: str, fix: str) -> None:
|
|
gaps.append({"key": key, "gap": gap, "why_it_matters": why, "suggested_fix": fix})
|
|
|
|
if fallback_count > 0:
|
|
add(
|
|
"citizen_generation",
|
|
f"{fallback_count} citizen(s) entered through degraded generation.",
|
|
"Their initial identity fields may be less representative than normally generated citizens.",
|
|
"Retry cast generation and inspect the affected residents before treating their behavior as evidence.",
|
|
)
|
|
|
|
if is_market and all(abs(v) < 1e-9 for v in values_by_key["adoption_rate"]):
|
|
add(
|
|
"adoption_rate",
|
|
"Adoption rate stayed at 0.00 for the whole run.",
|
|
"The report cannot infer an adoption trajectory from this simulation.",
|
|
"Increase the horizon or verify that purchase/proposal actions can occur.",
|
|
)
|
|
if "DEFECT" not in action_types and "ABANDON" not in action_types:
|
|
add(
|
|
"defection_or_abandonment",
|
|
"No DEFECT or ABANDON action occurred.",
|
|
"Churn, defection, and trust-loss conclusions are weakly evidenced.",
|
|
"Run a longer or more disruptive scenario, or report this as a limitation.",
|
|
)
|
|
if "PROPOSE_RULE" not in action_types and not is_market:
|
|
add(
|
|
"rule_evolution",
|
|
"No PROPOSE_RULE action occurred.",
|
|
"Institutional or rule-evolution claims are not supported by observed actions.",
|
|
"Include a rule-change pressure or extend the simulation horizon.",
|
|
)
|
|
if tension_event_count > 0:
|
|
add(
|
|
"tension_events",
|
|
f"{tension_event_count} tension event(s) occurred but are not represented as a first-class report section.",
|
|
"Readers may miss event-driven causes of later metric changes.",
|
|
"Include event descriptions and affected days in the report timeline.",
|
|
)
|
|
|
|
flat_keys = ["trust", "conflict", "information_spread"]
|
|
if is_market:
|
|
flat_keys.append("purchase_intent")
|
|
for key in flat_keys:
|
|
values = values_by_key[key]
|
|
if len(values) >= 3 and max(values) - min(values) < 0.005:
|
|
add(
|
|
key,
|
|
f"{METRIC_LABELS[key]} stayed effectively flat ({values[0]:.2f} → {values[-1]:.2f}).",
|
|
"A flat series does not show whether the mechanism is stable or simply inactive at this horizon.",
|
|
"Inspect the triggering actions and compare with a longer or counterfactual run.",
|
|
)
|
|
|
|
return gaps
|
|
|
|
@staticmethod
|
|
def compute_agent_stats(agents: list[AgentPersona], actions: list[ActionEntry]) -> list[dict]:
|
|
action_counts: dict[int, dict[str, int]] = defaultdict(lambda: defaultdict(int))
|
|
speeches: dict[int, list[str]] = defaultdict(list)
|
|
thoughts: dict[int, list[str]] = defaultdict(list)
|
|
|
|
for a in actions:
|
|
action_counts[a.agent_id][a.action_type.value] += 1
|
|
if a.speech:
|
|
speeches[a.agent_id].append(a.speech)
|
|
if a.internal_thought:
|
|
thoughts[a.agent_id].append(a.internal_thought)
|
|
|
|
results = []
|
|
for agent in agents:
|
|
results.append({
|
|
"id": agent.id,
|
|
"name": agent.name,
|
|
"role": agent.role,
|
|
"faction": agent.faction,
|
|
"emotional_state": agent.emotional_state,
|
|
"personality": agent.personality.model_dump(),
|
|
"action_counts": dict(action_counts.get(agent.id, {})),
|
|
"speeches": speeches.get(agent.id, [])[-10:],
|
|
"thoughts": thoughts.get(agent.id, [])[-10:],
|
|
"total_actions": sum(action_counts.get(agent.id, {}).values()),
|
|
})
|
|
return results
|
|
|
|
@staticmethod
|
|
def compute_segment_stats(agent_stats: list[dict]) -> list[dict]:
|
|
segments: dict[str, list[dict]] = defaultdict(list)
|
|
for a in agent_stats:
|
|
seg_key = a["role"] or "General"
|
|
segments[seg_key].append(a)
|
|
|
|
results = []
|
|
for seg_name, seg_agents in segments.items():
|
|
total = len(seg_agents)
|
|
purchased = sum(1 for a in seg_agents if a["action_counts"].get("PURCHASE", 0) > 0)
|
|
abandoned = sum(1 for a in seg_agents if a["action_counts"].get("ABANDON", 0) > 0)
|
|
recommended = sum(1 for a in seg_agents if a["action_counts"].get("RECOMMEND", 0) > 0)
|
|
compared = sum(1 for a in seg_agents if a["action_counts"].get("COMPARE", 0) > 0)
|
|
|
|
adopted = purchased - abandoned if purchased > abandoned else 0
|
|
churned = abandoned
|
|
|
|
results.append({
|
|
"name": seg_name,
|
|
"agent_count": total,
|
|
"funnel": {
|
|
"aware": total,
|
|
"interested": purchased + compared + recommended,
|
|
"tried": purchased,
|
|
"adopted": adopted,
|
|
"churned": churned,
|
|
},
|
|
"adoption_pct": round((adopted / total) * 100) if total > 0 else 0,
|
|
"advocate_count": recommended,
|
|
"top_speeches": [],
|
|
"top_thoughts": [],
|
|
})
|
|
|
|
all_speeches = []
|
|
all_thoughts = []
|
|
for a in seg_agents:
|
|
all_speeches.extend(a["speeches"][-3:])
|
|
all_thoughts.extend(a["thoughts"][-3:])
|
|
results[-1]["top_speeches"] = all_speeches[-15:]
|
|
results[-1]["top_thoughts"] = all_thoughts[-15:]
|
|
|
|
return results
|
|
|
|
@staticmethod
|
|
def detect_inflection_points(metrics_history: list[dict], threshold: float = 0.05) -> list[dict]:
|
|
if len(metrics_history) < 2:
|
|
return []
|
|
|
|
points = []
|
|
metric_keys = list(METRIC_LABELS.keys())
|
|
|
|
for i in range(1, len(metrics_history)):
|
|
curr = metrics_history[i]
|
|
prev = metrics_history[i - 1]
|
|
round_num = curr.get("round", i)
|
|
day = round_num // 3 + 1
|
|
|
|
for key in metric_keys:
|
|
delta = abs(curr.get(key, 0.5) - prev.get(key, 0.5))
|
|
if delta >= threshold:
|
|
direction = "spike" if curr.get(key, 0.5) > prev.get(key, 0.5) else "drop"
|
|
points.append({
|
|
"day": day,
|
|
"round": round_num,
|
|
"metric": key,
|
|
"label": METRIC_LABELS[key],
|
|
"delta": round(delta, 4),
|
|
"direction": direction,
|
|
"value_before": round(prev.get(key, 0.5), 4),
|
|
"value_after": round(curr.get(key, 0.5), 4),
|
|
})
|
|
|
|
points.sort(key=lambda x: x["delta"], reverse=True)
|
|
return points[:20]
|
|
|
|
@staticmethod
|
|
def compute_confidence(metrics_history: list[dict], agent_count: int) -> str:
|
|
rounds = len(metrics_history)
|
|
if rounds >= 30 and agent_count >= 15:
|
|
return "high"
|
|
if rounds >= 15 and agent_count >= 8:
|
|
return "medium"
|
|
return "low"
|
|
|
|
@staticmethod
|
|
def format_causal_links(causal_links: list) -> list[dict]:
|
|
"""Format CausalLink objects for the report."""
|
|
return [
|
|
{
|
|
"cause": l.cause,
|
|
"cause_label": l.cause.replace("_", " ").title(),
|
|
"effect": l.effect,
|
|
"effect_label": l.effect.replace("_", " ").title(),
|
|
"lag": l.lag,
|
|
"p_value": l.p_value,
|
|
"strength": l.strength,
|
|
"description": (
|
|
f"{l.cause.replace('_', ' ').title()} Granger-causes "
|
|
f"{l.effect.replace('_', ' ').title()} with a {l.lag}-round lag "
|
|
f"(p={l.p_value:.3f}, {l.strength} evidence)"
|
|
),
|
|
}
|
|
for l in causal_links
|
|
]
|
|
|
|
@staticmethod
|
|
def format_counterfactuals(counterfactuals: list) -> list[dict]:
|
|
"""Format CounterfactualResult objects for the report."""
|
|
results = []
|
|
for c in counterfactuals:
|
|
significant_impacts = {
|
|
k: v for k, v in c.metric_impacts.items() if abs(v) > 0.03
|
|
}
|
|
if not significant_impacts:
|
|
continue
|
|
results.append({
|
|
"event_round": c.event_round,
|
|
"event_description": c.event_description,
|
|
"impacts": [
|
|
{
|
|
"metric": k,
|
|
"label": k.replace("_", " ").title(),
|
|
"delta": round(v, 3),
|
|
"direction": "increased" if v > 0 else "decreased",
|
|
"description": (
|
|
f"{k.replace('_', ' ').title()} {'increased' if v > 0 else 'decreased'} "
|
|
f"by {abs(v):.2f} due to this event"
|
|
),
|
|
}
|
|
for k, v in sorted(significant_impacts.items(), key=lambda x: abs(x[1]), reverse=True)
|
|
],
|
|
})
|
|
return results
|
|
|
|
@staticmethod
|
|
def format_coherence_scores(coherence_scores: dict) -> dict:
|
|
"""Summarize agent coherence for the report."""
|
|
if not coherence_scores:
|
|
return {"overall_score": 1.0, "total_agents": 0, "coherent_count": 0, "flagged_agents": []}
|
|
|
|
scores = list(coherence_scores.values())
|
|
overall = sum(s.score for s in scores) / len(scores) if scores else 1.0
|
|
coherent = sum(1 for s in scores if s.score >= 0.5)
|
|
flagged = [
|
|
{
|
|
"agent_id": s.agent_id,
|
|
"score": round(s.score, 2),
|
|
"contradictions": s.contradictions,
|
|
}
|
|
for s in sorted(scores, key=lambda x: x.score)
|
|
if s.score < 0.5
|
|
][:5]
|
|
|
|
return {
|
|
"overall_score": round(overall, 2),
|
|
"total_agents": len(scores),
|
|
"coherent_count": coherent,
|
|
"coherent_pct": round((coherent / len(scores)) * 100) if scores else 100,
|
|
"flagged_agents": flagged,
|
|
}
|