references/feedback-integration.md
# Feedback Integration Guide
## Overview
This document explains how signal postmortem results integrate with downstream systems: edge-signal-aggregator weight calibration and the skill improvement backlog.
## Feedback Targets
### 1. Edge-Signal-Aggregator Weight Calibration
**Purpose**: Adjust the weight of each contributing skill based on historical accuracy.
**Integration Point**: `reports/weight_feedback_YYYY-MM-DD.json`
**Consumption**:
1. edge-signal-aggregator reads feedback file on startup
2. Applies weight adjustments to skill contribution scores
3. Logs adjustment application for audit trail
**Weight Adjustment Formula**:
```
new_weight = current_weight * (1 + adjustment_factor)
adjustment_factor = (accuracy - baseline) * sensitivity
where:
- accuracy = true_positives / (true_positives + false_positives)
- baseline = 0.55 (expected random accuracy)
- sensitivity = 0.5 (dampening factor to avoid overreaction)
```
**Constraints**:
- Minimum weight: 0.3 (never fully disable a skill)
- Maximum weight: 2.0 (never over-amplify)
- Minimum sample size: 20 signals for adjustment
- Rolling window: 90 days of postmortem data
### 2. Skill Improvement Backlog
**Purpose**: Generate actionable improvement tasks for the skill improvement loop.
**Integration Point**: `reports/skill_improvement_backlog.yaml`
**Consumption**:
1. Skill improvement loop reads backlog entries
2. Prioritizes by severity and sample size
3. Creates improvement branches for high-severity issues
**Issue Types**:
| Issue Type | Trigger | Severity |
|------------|---------|----------|
| `false_positive_cluster` | >15% FP rate with 20+ samples | MEDIUM-HIGH |
| `regime_sensitivity` | >25% regime mismatch rate | MEDIUM |
| `sector_blind_spot` | >20% FP rate in specific sector | MEDIUM |
| `timing_drift` | Accuracy degraded >10% over 30 days | LOW-MEDIUM |
| `overconfidence` | High-confidence signals underperforming | HIGH |
**Backlog Entry Format**:
```yaml
- skill: vcp-screener
issue_type: false_positive_cluster
severity: medium
evidence:
false_positive_rate: 0.18
sample_size: 45
regime_correlation: RISK_OFF
sector_correlation: Technology
suggested_action: "Add RISK_OFF regime filter or reduce confidence"
priority_score: 72 # Calculated from severity * sample_size * impact
generated_by: signal-postmortem
generated_at: "2026-03-17T10:35:00Z"
status: pending
```
## Feedback Frequency
### Real-Time (Per Signal)
- Postmortem record created immediately after trade closure
- Stored in `reports/postmortems/`
### Daily Batch
- Weight feedback regenerated daily at 06:00
- Improvement backlog updated with new entries
- Old entries marked as `addressed` when improvements deployed
### Weekly Review
- Summary statistics by skill, sector, regime
- Trend analysis (rolling 4-week accuracy)
- Human review flagged for significant changes
## Data Flow Diagram
```
Signal Generated
|
v
Trade Executed (or skipped)
|
v
Holding Period Completes
|
v
postmortem_recorder.py
|
+---> postmortem record (JSON)
|
v
postmortem_analyzer.py
|
+---> weight_feedback.json --> edge-signal-aggregator
|
+---> skill_improvement_backlog.yaml --> skill improvement loop
|
+---> summary report (Markdown)
```
## Idempotency and Deduplication
### Postmortem Records
- `postmortem_id` = `pm_` + `signal_id`
- If postmortem already exists, update instead of duplicate
- Version field tracks updates
### Weight Feedback
- Regenerated fresh each run (not cumulative)
- Based on rolling 90-day window
- Old feedback files archived to `reports/archive/`
### Backlog Entries
- Keyed by `skill` + `issue_type` + `month`
- New evidence updates existing entry instead of creating duplicate
- Entry moved to `addressed` when skill version changes
## Minimum Thresholds
To avoid noisy feedback from small samples:
| Metric | Minimum |
|--------|---------|
| Weight adjustment | 20 signals |
| Backlog entry | 15 signals |
| Summary statistics | 10 signals |
| Regime correlation | 10 signals per regime |
## Manual Override Integration
When human review identifies an issue not caught by automated analysis:
```yaml
# Manual backlog entry
- skill: earnings-trade-analyzer
issue_type: manual_review
severity: high
evidence:
description: "Systematic gap fade failure in biotech earnings"
reviewer: "tradermonty"
suggested_action: "Exclude biotech from gap fade signals"
generated_by: human_review
generated_at: "2026-03-17T14:00:00Z"
```
Manual entries have `generated_by: human_review` and are prioritized higher.
## Conflict Resolution
When automated and manual feedback conflict:
1. Manual feedback takes precedence for immediate action
2. Automated feedback triggers investigation
3. Resolution documented in backlog entry
## Audit Trail
All feedback actions are logged:
```json
{
"action": "weight_adjustment_applied",
"skill": "vcp-screener",
"old_weight": 1.0,
"new_weight": 0.85,
"reason": "15% FP rate in RISK_OFF",
"applied_at": "2026-03-17T06:00:00Z",
"applied_by": "edge-signal-aggregator"
}
```
Logs stored in `logs/feedback_audit.log` with 90-day retention.
references/outcome-classification.md
# Outcome Classification Guide
## Overview
This document defines how signal outcomes are classified in the postmortem process. Accurate classification is essential for meaningful feedback to edge-signal-aggregator and skill improvement.
## Classification Categories
### 1. TRUE_POSITIVE
**Definition**: The predicted direction matched the realized return sign.
**Criteria**:
- LONG signal with positive realized return at holding period
- SHORT signal with negative realized return at holding period
- Minimum threshold: |return| >= 0.5% to avoid noise classification
**Examples**:
- Signal: LONG AAPL at $170, predicted 5-day upside
- Outcome: AAPL at $175 after 5 days (+2.9%)
- Classification: TRUE_POSITIVE
### 2. FALSE_POSITIVE
**Definition**: The predicted direction was opposite to the realized return.
**Criteria**:
- LONG signal with negative realized return at holding period
- SHORT signal with positive realized return at holding period
- Minimum threshold: |return| >= 0.5% to avoid noise classification
**Sub-categories**:
- `FALSE_POSITIVE_MILD`: -0.5% to -2% for LONG (or +0.5% to +2% for SHORT)
- `FALSE_POSITIVE_SEVERE`: worse than -2% for LONG (or +2% for SHORT)
**Examples**:
- Signal: LONG NVDA at $900, predicted breakout
- Outcome: NVDA at $870 after 5 days (-3.3%)
- Classification: FALSE_POSITIVE_SEVERE
### 3. MISSED_OPPORTUNITY
**Definition**: A signal that was generated but not acted upon, and would have been profitable.
**Use Cases**:
- Signals filtered out by aggregator confidence threshold
- Signals skipped due to position sizing constraints
- Signals in watchlist but not traded
**Criteria**:
- Signal was generated but `trade_taken = false`
- Realized return in predicted direction >= 2%
**Note**: This category helps identify overly conservative filtering.
### 4. REGIME_MISMATCH
**Definition**: Signal failed primarily due to a market regime change rather than skill error.
**Criteria**:
- `regime_at_signal` differs from `regime_at_exit`
- AND return in opposite direction of prediction
- Regime change must be documented (e.g., VIX spike, Fed announcement)
**Examples**:
- Signal: LONG growth stock on 2026-03-05 (RISK_ON regime)
- Outcome: Tariff announcement on 2026-03-07 triggered RISK_OFF
- Result: Stock down 8% due to sector rotation
- Classification: REGIME_MISMATCH (not FALSE_POSITIVE)
**Regime Detection**:
- RISK_ON: VIX < 20, breadth > 60%, leading stocks advancing
- RISK_OFF: VIX > 25, breadth < 40%, defensive rotation
- TRANSITION: Mixed signals, high uncertainty
## Edge Cases
### Flat Outcome (|return| < 0.5%)
- Classification: `NEUTRAL`
- Does not count as TRUE_POSITIVE or FALSE_POSITIVE
- May indicate weak signal strength
### Early Exit
When a trade is closed before the target holding period:
```
holding_period_actual < holding_period_target
```
- Use actual holding period for return calculation
- Note `early_exit = true` in postmortem record
- Include `early_exit_reason`: stop_loss, target_reached, discretionary
### Gap Events
If the stock gapped significantly at open due to overnight news:
- Record `gap_event = true`
- Include `gap_pct` in postmortem
- Consider separate analysis for gap-driven outcomes
### Multiple Holding Periods
The skill tracks both 5-day and 20-day returns:
| Metric | 5-Day | 20-Day |
|--------|-------|--------|
| Purpose | Short-term edge validation | Medium-term edge validation |
| Threshold | 0.5% | 1.0% |
| Weight for feedback | 60% | 40% |
A signal can be TRUE_POSITIVE at 5 days but FALSE_POSITIVE at 20 days (or vice versa). Both are recorded.
## Classification Decision Tree
```
1. Was the trade taken?
NO -> If would have been profitable: MISSED_OPPORTUNITY
Otherwise: SKIPPED (no postmortem needed)
YES -> Continue
2. Did regime change during holding period?
YES -> Is return in wrong direction AND > 2% loss?
YES -> REGIME_MISMATCH
NO -> Continue
NO -> Continue
3. Is |return| < 0.5%?
YES -> NEUTRAL
NO -> Continue
4. Does return sign match predicted direction?
YES -> TRUE_POSITIVE
NO -> FALSE_POSITIVE (check severity)
```
## Attribution Rules
### Single-Source Signals
When a signal comes from one skill (e.g., vcp-screener alone):
- Full attribution to that skill
### Aggregated Signals
When a signal comes from edge-signal-aggregator combining multiple skills:
- Attribution proportional to each skill's contribution weight
- Example: VCP (0.4) + CANSLIM (0.3) + Breadth (0.3)
- If FALSE_POSITIVE: each skill receives proportional negative feedback
### Override Signals
When a human overrides a skill recommendation:
- Record `human_override = true`
- Separate analysis track for human decision quality
## Confidence Adjustment Factors
Classification confidence is adjusted based on:
| Factor | Adjustment |
|--------|------------|
| High volume day | +10% confidence |
| Low volume day | -10% confidence |
| Earnings during holding period | -20% confidence |
| VIX spike > 5 points | -15% confidence |
| Large gap (> 3%) | -15% confidence |
Final confidence is capped at [0.5, 1.0] range.
requirements.txt
# Runtime dependencies for standalone .skill installs (issue #330).
# Verified by scripts/check_skill_deps.py.
requests>=2.31.0
scripts/postmortem_analyzer.py
#!/usr/bin/env python3
"""
Signal Postmortem Analyzer
Generates feedback for edge-signal-aggregator weight calibration
and skill improvement backlog entries from postmortem records.
"""
from __future__ import annotations
import argparse
import json
import sys
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from pathlib import Path
def load_postmortems(
postmortems_dir: str, days_back: int = 90, now: datetime | None = None
) -> list:
"""
Load postmortem records from directory.
Args:
postmortems_dir: Directory containing postmortem JSON files
days_back: Only include postmortems from this many days back
now: Optional fixed reference time for the cutoff window. When omitted,
wall-clock time is used. Passing a fixed anchor keeps the window
deterministic for replay harnesses whose fixtures are dated
independently of the run date.
Returns:
List of postmortem records
"""
postmortems = []
dir_path = Path(postmortems_dir)
if not dir_path.exists():
return postmortems
ref_time = (now or datetime.now()).replace(tzinfo=None)
cutoff_date = ref_time - timedelta(days=days_back)
for json_file in dir_path.glob("pm_*.json"):
try:
with open(json_file) as f:
pm = json.load(f)
# Check date
recorded_at = pm.get("recorded_at", "")
if recorded_at:
try:
rec_dt = datetime.fromisoformat(recorded_at.replace("Z", "+00:00"))
if rec_dt.replace(tzinfo=None) < cutoff_date:
continue
except ValueError:
pass
postmortems.append(pm)
except (OSError, json.JSONDecodeError) as e:
print(f"Warning: Error loading {json_file}: {e}", file=sys.stderr)
return postmortems
def calculate_skill_metrics(postmortems: list) -> dict:
"""
Calculate accuracy and other metrics by source skill.
Returns dict mapping skill name to metrics.
"""
skill_data = defaultdict(
lambda: {
"true_positive": 0,
"false_positive": 0,
"false_positive_severe": 0,
"regime_mismatch": 0,
"neutral": 0,
"total": 0,
"total_return_5d": 0.0,
"returns": [],
}
)
for pm in postmortems:
skill = pm.get("source_skill", "unknown")
outcome = pm.get("outcome_category", "UNKNOWN")
returns_5d = pm.get("realized_returns", {}).get("5d", 0.0)
skill_data[skill]["total"] += 1
skill_data[skill]["total_return_5d"] += returns_5d
skill_data[skill]["returns"].append(returns_5d)
if outcome == "TRUE_POSITIVE":
skill_data[skill]["true_positive"] += 1
elif outcome in ("FALSE_POSITIVE", "FALSE_POSITIVE_SEVERE"):
skill_data[skill]["false_positive"] += 1
if outcome == "FALSE_POSITIVE_SEVERE":
skill_data[skill]["false_positive_severe"] += 1
elif outcome == "REGIME_MISMATCH":
skill_data[skill]["regime_mismatch"] += 1
elif outcome == "NEUTRAL":
skill_data[skill]["neutral"] += 1
# Calculate derived metrics
metrics = {}
for skill, data in skill_data.items():
total_decisions = data["true_positive"] + data["false_positive"]
accuracy = data["true_positive"] / total_decisions if total_decisions > 0 else 0.0
fp_rate = data["false_positive"] / data["total"] if data["total"] > 0 else 0.0
avg_return = data["total_return_5d"] / data["total"] if data["total"] > 0 else 0.0
metrics[skill] = {
"sample_size": data["total"],
"true_positive": data["true_positive"],
"false_positive": data["false_positive"],
"false_positive_severe": data["false_positive_severe"],
"regime_mismatch": data["regime_mismatch"],
"neutral": data["neutral"],
"accuracy": accuracy,
"false_positive_rate": fp_rate,
"avg_return_5d": avg_return,
}
return metrics
def generate_weight_feedback(
metrics: dict,
min_sample_size: int = 20,
baseline_accuracy: float = 0.55,
sensitivity: float = 0.5,
) -> dict:
"""
Generate weight adjustment suggestions for edge-signal-aggregator.
"""
adjustments = []
for skill, data in metrics.items():
if data["sample_size"] < min_sample_size:
continue
accuracy = data["accuracy"]
adjustment_factor = (accuracy - baseline_accuracy) * sensitivity
# Clamp to reasonable range
new_weight = max(0.3, min(2.0, 1.0 + adjustment_factor))
# Only suggest if meaningful change
if abs(1.0 - new_weight) >= 0.05:
reason = []
if data["false_positive_rate"] > 0.15:
reason.append(f"{data['false_positive_rate']:.0%} false positive rate")
if accuracy < baseline_accuracy:
reason.append(
f"below baseline accuracy ({accuracy:.0%} vs {baseline_accuracy:.0%})"
)
if accuracy > 0.65:
reason.append(f"strong accuracy ({accuracy:.0%})")
adjustments.append(
{
"skill": skill,
"current_weight": 1.0,
"suggested_weight": round(new_weight, 2),
"reason": "; ".join(reason) if reason else "accuracy-based adjustment",
"sample_size": data["sample_size"],
"accuracy": round(accuracy, 3),
"false_positive_rate": round(data["false_positive_rate"], 3),
}
)
# Determine confidence based on total sample size
total_samples = sum(m["sample_size"] for m in metrics.values())
if total_samples >= 100:
confidence = "HIGH"
elif total_samples >= 50:
confidence = "MEDIUM"
else:
confidence = "LOW"
return {
"schema_version": "1.0",
"generated_at": datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + "Z",
"analysis_period": {"days_back": 90, "total_postmortems": total_samples},
"skill_adjustments": adjustments,
"confidence": confidence,
"min_sample_threshold": min_sample_size,
}
def generate_improvement_backlog(
metrics: dict, postmortems: list, min_sample_size: int = 15
) -> list:
"""
Generate skill improvement backlog entries.
"""
entries = []
now = datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + "Z"
for skill, data in metrics.items():
if data["sample_size"] < min_sample_size:
continue
# Check for false positive cluster
if data["false_positive_rate"] > 0.15:
# Analyze regime correlation
regime_correlation = analyze_regime_correlation(postmortems, skill)
severity = "high" if data["false_positive_rate"] > 0.25 else "medium"
priority_score = int(data["false_positive_rate"] * data["sample_size"] * 100)
entries.append(
{
"skill": skill,
"issue_type": "false_positive_cluster",
"severity": severity,
"evidence": {
"false_positive_rate": round(data["false_positive_rate"], 3),
"sample_size": data["sample_size"],
"regime_correlation": regime_correlation,
},
"suggested_action": f"Add regime filter or reduce confidence in {regime_correlation} regime"
if regime_correlation != "NONE"
else "Review signal generation logic for systematic errors",
"priority_score": priority_score,
"generated_by": "signal-postmortem",
"generated_at": now,
"status": "pending",
}
)
# Check for regime sensitivity
regime_mismatch_rate = (
data["regime_mismatch"] / data["sample_size"] if data["sample_size"] > 0 else 0
)
if regime_mismatch_rate > 0.25:
entries.append(
{
"skill": skill,
"issue_type": "regime_sensitivity",
"severity": "medium",
"evidence": {
"regime_mismatch_rate": round(regime_mismatch_rate, 3),
"sample_size": data["sample_size"],
},
"suggested_action": "Incorporate market regime detection into signal generation",
"priority_score": int(regime_mismatch_rate * data["sample_size"] * 80),
"generated_by": "signal-postmortem",
"generated_at": now,
"status": "pending",
}
)
# Check for severe false positives (overconfidence)
severe_fp_rate = (
data["false_positive_severe"] / data["sample_size"] if data["sample_size"] > 0 else 0
)
if severe_fp_rate > 0.10:
entries.append(
{
"skill": skill,
"issue_type": "overconfidence",
"severity": "high",
"evidence": {
"severe_fp_rate": round(severe_fp_rate, 3),
"severe_fp_count": data["false_positive_severe"],
"sample_size": data["sample_size"],
},
"suggested_action": "Review confidence scoring; reduce position size or add confirmation filters",
"priority_score": int(severe_fp_rate * data["sample_size"] * 150),
"generated_by": "signal-postmortem",
"generated_at": now,
"status": "pending",
}
)
# Sort by priority score descending
entries.sort(key=lambda x: x["priority_score"], reverse=True)
return entries
def analyze_regime_correlation(postmortems: list, skill: str) -> str:
"""
Analyze which regime has the most false positives for a skill.
"""
regime_fps = defaultdict(int)
regime_totals = defaultdict(int)
for pm in postmortems:
if pm.get("source_skill") != skill:
continue
regime = pm.get("regime_at_signal", "UNKNOWN")
regime_totals[regime] += 1
if pm.get("outcome_category", "").startswith("FALSE_POSITIVE"):
regime_fps[regime] += 1
# Find regime with highest FP rate
worst_regime = "NONE"
worst_rate = 0.0
for regime, total in regime_totals.items():
if total < 5: # Minimum for correlation
continue
rate = regime_fps[regime] / total
if rate > worst_rate and rate > 0.2: # Threshold for correlation
worst_rate = rate
worst_regime = regime
return worst_regime
def generate_summary(metrics: dict, postmortems: list, group_by: list) -> str:
"""
Generate markdown summary report.
"""
lines = [
"# Signal Postmortem Summary",
"",
f"**Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
f"**Total Postmortems:** {len(postmortems)}",
"",
]
# Overall statistics
total_tp = sum(m["true_positive"] for m in metrics.values())
total_fp = sum(m["false_positive"] for m in metrics.values())
total_all = sum(m["sample_size"] for m in metrics.values())
overall_accuracy = total_tp / (total_tp + total_fp) if (total_tp + total_fp) > 0 else 0
lines.extend(
[
"## Overall Statistics",
"",
"| Metric | Value |",
"|--------|-------|",
f"| Total Signals | {total_all} |",
f"| True Positives | {total_tp} |",
f"| False Positives | {total_fp} |",
f"| Overall Accuracy | {overall_accuracy:.1%} |",
"",
]
)
# By skill breakdown
if "skill" in group_by:
lines.extend(
[
"## By Skill",
"",
"| Skill | Samples | Accuracy | FP Rate | Avg Return (5d) |",
"|-------|---------|----------|---------|-----------------|",
]
)
for skill, data in sorted(metrics.items(), key=lambda x: x[1]["sample_size"], reverse=True):
lines.append(
f"| {skill} | {data['sample_size']} | {data['accuracy']:.1%} | {data['false_positive_rate']:.1%} | {data['avg_return_5d']:.2%} |"
)
lines.append("")
# By month breakdown
if "month" in group_by:
monthly = defaultdict(lambda: {"tp": 0, "fp": 0, "total": 0})
for pm in postmortems:
signal_date = pm.get("signal_date", "")
if not signal_date:
continue
month_key = signal_date[:7] # YYYY-MM
monthly[month_key]["total"] += 1
outcome = pm.get("outcome_category", "")
if outcome == "TRUE_POSITIVE":
monthly[month_key]["tp"] += 1
elif outcome.startswith("FALSE_POSITIVE"):
monthly[month_key]["fp"] += 1
lines.extend(
[
"## By Month",
"",
"| Month | Signals | Accuracy |",
"|-------|---------|----------|",
]
)
for month, data in sorted(monthly.items()):
decisions = data["tp"] + data["fp"]
acc = data["tp"] / decisions if decisions > 0 else 0
lines.append(f"| {month} | {data['total']} | {acc:.1%} |")
lines.append("")
# Outcome distribution
outcome_dist = defaultdict(int)
for pm in postmortems:
outcome = pm.get("outcome_category", "UNKNOWN")
outcome_dist[outcome] += 1
lines.extend(
[
"## Outcome Distribution",
"",
"| Outcome | Count | Percentage |",
"|---------|-------|------------|",
]
)
for outcome, count in sorted(outcome_dist.items(), key=lambda x: x[1], reverse=True):
pct = count / len(postmortems) if postmortems else 0
lines.append(f"| {outcome} | {count} | {pct:.1%} |")
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="Analyze postmortem records and generate feedback")
parser.add_argument(
"--postmortems-dir",
default="reports/postmortems/",
help="Directory containing postmortem JSON files (default: reports/postmortems/)",
)
parser.add_argument(
"--days-back",
type=int,
default=90,
help="Analyze postmortems from this many days back (default: 90)",
)
parser.add_argument(
"--generate-weight-feedback",
action="store_true",
help="Generate weight adjustment suggestions for edge-signal-aggregator",
)
parser.add_argument(
"--generate-improvement-backlog",
action="store_true",
help="Generate skill improvement backlog entries",
)
parser.add_argument("--summary", action="store_true", help="Generate summary statistics report")
parser.add_argument(
"--group-by",
default="skill,month",
help="Comma-separated grouping for summary (default: skill,month)",
)
parser.add_argument(
"--min-sample-size",
type=int,
default=20,
help="Minimum samples for weight feedback (default: 20)",
)
parser.add_argument(
"--output-dir", default="reports/", help="Output directory (default: reports/)"
)
args = parser.parse_args()
# Load postmortems
postmortems = load_postmortems(args.postmortems_dir, args.days_back)
if not postmortems:
print(f"No postmortems found in {args.postmortems_dir}", file=sys.stderr)
sys.exit(1)
print(f"Loaded {len(postmortems)} postmortems")
# Calculate metrics
metrics = calculate_skill_metrics(postmortems)
# Create output directory
output_dir = Path(args.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y-%m-%d")
# Generate weight feedback
if args.generate_weight_feedback:
feedback = generate_weight_feedback(metrics, args.min_sample_size)
output_file = output_dir / f"weight_feedback_{timestamp}.json"
with open(output_file, "w", encoding="utf-8") as f:
json.dump(feedback, f, indent=2)
print(f"Saved weight feedback: {output_file}")
print(f" Adjustments: {len(feedback['skill_adjustments'])}")
print(f" Confidence: {feedback['confidence']}")
# Generate improvement backlog
if args.generate_improvement_backlog:
backlog = generate_improvement_backlog(metrics, postmortems, min(15, args.min_sample_size))
output_file = output_dir / f"skill_improvement_backlog_{timestamp}.yaml"
# Write as YAML-like format
with open(output_file, "w", encoding="utf-8") as f:
for entry in backlog:
f.write(f"- skill: {entry['skill']}\n")
f.write(f" issue_type: {entry['issue_type']}\n")
f.write(f" severity: {entry['severity']}\n")
f.write(" evidence:\n")
for k, v in entry["evidence"].items():
f.write(f" {k}: {v}\n")
f.write(f' suggested_action: "{entry["suggested_action"]}"\n')
f.write(f" priority_score: {entry['priority_score']}\n")
f.write(f" generated_by: {entry['generated_by']}\n")
f.write(f' generated_at: "{entry["generated_at"]}"\n')
f.write(f" status: {entry['status']}\n")
f.write("\n")
print(f"Saved improvement backlog: {output_file}")
print(f" Entries: {len(backlog)}")
# Generate summary
if args.summary:
group_by = [g.strip() for g in args.group_by.split(",")]
summary = generate_summary(metrics, postmortems, group_by)
output_file = output_dir / f"postmortem_summary_{timestamp}.md"
with open(output_file, "w", encoding="utf-8") as f:
f.write(summary)
print(f"Saved summary: {output_file}")
# Default: print metrics if no specific output requested
if not (args.generate_weight_feedback or args.generate_improvement_backlog or args.summary):
print("\nMetrics by skill:")
for skill, data in sorted(metrics.items(), key=lambda x: x[1]["sample_size"], reverse=True):
print(f"\n{skill}:")
print(f" Samples: {data['sample_size']}")
print(f" Accuracy: {data['accuracy']:.1%}")
print(f" FP Rate: {data['false_positive_rate']:.1%}")
print(f" Avg Return (5d): {data['avg_return_5d']:.2%}")
if __name__ == "__main__":
main()
scripts/postmortem_recorder.py
#!/usr/bin/env python3
"""
Signal Postmortem Recorder
Records post-trade outcomes for signals generated by edge pipeline and other skills.
Fetches realized returns (5-day, 20-day) and classifies outcomes.
"""
import argparse
import json
import os
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Optional
# For FMP API calls
try:
import requests
HAS_REQUESTS = True
except ImportError:
HAS_REQUESTS = False
def get_fmp_api_key() -> Optional[str]:
"""Get FMP API key from environment or return None."""
return os.environ.get("FMP_API_KEY")
def fetch_price_data(ticker: str, start_date: str, end_date: str, api_key: str) -> dict:
"""
Fetch historical price data from FMP API (stable with v3 fallback).
Returns dict mapping date string to close price.
"""
if not HAS_REQUESTS:
return {}
endpoints = [
("https://financialmodelingprep.com/stable/historical-price-eod/full", True),
("https://financialmodelingprep.com/api/v3/historical-price-full", False),
]
for base_url, is_stable in endpoints:
try:
if is_stable:
url = base_url
params = {"symbol": ticker, "from": start_date, "to": end_date, "apikey": api_key}
else:
url = f"{base_url}/{ticker}"
params = {"from": start_date, "to": end_date, "apikey": api_key}
resp = requests.get(url, params=params, timeout=30)
if resp.status_code != 200:
continue
data = resp.json()
historical = None
if isinstance(data, dict) and "historical" in data:
historical = data["historical"]
elif isinstance(data, dict) and "historicalStockList" in data:
for entry in data["historicalStockList"]:
if entry.get("symbol", "").replace("-", ".") == ticker.replace("-", "."):
historical = entry.get("historical", [])
break
if historical is not None:
return {item["date"]: item["close"] for item in historical}
except Exception: # nosec B112 - intentional fallback to next FMP endpoint
continue
print(
f"Warning: Failed to fetch price data for {ticker}: all endpoints failed", file=sys.stderr
)
return {}
def calculate_return(entry_price: float, exit_price: float) -> float:
"""Calculate percentage return."""
if entry_price <= 0:
return 0.0
return (exit_price - entry_price) / entry_price
def classify_outcome(
predicted_direction: str,
return_pct: float,
regime_at_signal: str,
regime_at_exit: str,
threshold: float = 0.005,
) -> str:
"""
Classify signal outcome based on realized return and regime context.
Returns one of: TRUE_POSITIVE, FALSE_POSITIVE, FALSE_POSITIVE_SEVERE,
REGIME_MISMATCH, NEUTRAL
"""
# Check for regime mismatch
if regime_at_signal != regime_at_exit and regime_at_exit != "UNKNOWN":
if predicted_direction == "LONG" and return_pct < -0.02:
return "REGIME_MISMATCH"
if predicted_direction == "SHORT" and return_pct > 0.02:
return "REGIME_MISMATCH"
# Check for neutral (flat) outcome
if abs(return_pct) < threshold:
return "NEUTRAL"
# Check direction match
direction_match = (predicted_direction == "LONG" and return_pct > 0) or (
predicted_direction == "SHORT" and return_pct < 0
)
if direction_match:
return "TRUE_POSITIVE"
# False positive - check severity
if abs(return_pct) > 0.02:
return "FALSE_POSITIVE_SEVERE"
return "FALSE_POSITIVE"
def create_postmortem_record(
signal: dict,
realized_returns: dict,
exit_price: float,
exit_date: str,
regime_at_exit: str = "UNKNOWN",
outcome_notes: str = "",
) -> dict:
"""
Create a postmortem record for a signal.
"""
signal_id = signal.get("signal_id", "unknown")
ticker = signal.get("ticker", "")
signal_date = signal.get("signal_date", "")
predicted_direction = signal.get("predicted_direction", "LONG")
source_skill = signal.get("source_skill", "unknown")
entry_price = signal.get("entry_price", 0.0)
regime_at_signal = signal.get("regime", "UNKNOWN")
# Use 5-day return for primary classification
primary_return = realized_returns.get("5d", 0.0)
outcome_category = classify_outcome(
predicted_direction, primary_return, regime_at_signal, regime_at_exit
)
# Calculate holding days
holding_days = 0
if signal_date and exit_date:
try:
sig_dt = datetime.strptime(signal_date, "%Y-%m-%d")
exit_dt = datetime.strptime(exit_date, "%Y-%m-%d")
holding_days = (exit_dt - sig_dt).days
except ValueError:
pass
return {
"schema_version": "1.0",
"postmortem_id": f"pm_{signal_id}",
"signal_id": signal_id,
"ticker": ticker,
"signal_date": signal_date,
"source_skill": source_skill,
"predicted_direction": predicted_direction,
"entry_price": entry_price,
"realized_returns": realized_returns,
"exit_price": exit_price,
"exit_date": exit_date,
"holding_days": holding_days,
"outcome_category": outcome_category,
"regime_at_signal": regime_at_signal,
"regime_at_exit": regime_at_exit,
"outcome_notes": outcome_notes,
"recorded_at": datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + "Z",
}
def process_signal(
signal: dict,
holding_periods: list,
api_key: Optional[str] = None,
manual_exit_price: Optional[float] = None,
manual_exit_date: Optional[str] = None,
) -> Optional[dict]:
"""
Process a single signal and create postmortem record.
"""
ticker = signal.get("ticker", "")
signal_date = signal.get("signal_date", "")
entry_price = signal.get("entry_price", 0.0)
if not ticker or not signal_date:
print(f"Warning: Signal missing ticker or date: {signal}", file=sys.stderr)
return None
# Calculate date range for price fetching
try:
sig_dt = datetime.strptime(signal_date, "%Y-%m-%d")
except ValueError:
print(f"Warning: Invalid signal date: {signal_date}", file=sys.stderr)
return None
max_period = max(holding_periods)
end_dt = sig_dt + timedelta(days=max_period + 5) # Buffer for weekends
# Fetch price data if API key available
prices = {}
if api_key:
prices = fetch_price_data(ticker, signal_date, end_dt.strftime("%Y-%m-%d"), api_key)
# Get entry price
if entry_price <= 0:
entry_price = prices.get(signal_date, 0.0)
if entry_price <= 0:
print(f"Warning: No entry price for {ticker} on {signal_date}", file=sys.stderr)
return None
# Calculate realized returns for each holding period
realized_returns = {}
exit_price = manual_exit_price
exit_date = manual_exit_date
for period in holding_periods:
target_dt = sig_dt + timedelta(days=period)
# Find closest trading day
for offset in range(5): # Check up to 5 days forward for weekends
check_date = (target_dt + timedelta(days=offset)).strftime("%Y-%m-%d")
if check_date in prices:
period_price = prices[check_date]
realized_returns[f"{period}d"] = calculate_return(entry_price, period_price)
# Use first period's exit as default
if exit_price is None:
exit_price = period_price
exit_date = check_date
break
# Use manual exit if provided
if manual_exit_price is not None:
exit_price = manual_exit_price
if manual_exit_date is not None:
exit_date = manual_exit_date
if exit_price is None:
exit_price = entry_price # No change if no data
if exit_date is None:
exit_date = (sig_dt + timedelta(days=holding_periods[0])).strftime("%Y-%m-%d")
return create_postmortem_record(signal, realized_returns, exit_price, exit_date)
def list_ready_signals(signals_dir: str, min_days: int = 5) -> list:
"""
List signals that are ready for postmortem (past minimum holding period).
"""
ready = []
signals_path = Path(signals_dir)
if not signals_path.exists():
return ready
cutoff_date = datetime.now() - timedelta(days=min_days)
for json_file in signals_path.glob("*.json"):
try:
with open(json_file) as f:
data = json.load(f)
signals = data if isinstance(data, list) else data.get("signals", [])
for signal in signals:
signal_date = signal.get("signal_date", "")
if signal_date:
sig_dt = datetime.strptime(signal_date, "%Y-%m-%d")
if sig_dt <= cutoff_date:
ready.append(signal)
except (json.JSONDecodeError, KeyError) as e:
print(f"Warning: Error reading {json_file}: {e}", file=sys.stderr)
return ready
def main():
parser = argparse.ArgumentParser(description="Record post-trade outcomes for signals")
parser.add_argument("--signals-file", help="JSON file containing signals to process")
parser.add_argument("--signals-dir", help="Directory containing signal JSON files")
parser.add_argument("--signal-id", help="Specific signal ID for manual recording")
parser.add_argument(
"--holding-periods",
default="5,20",
help="Comma-separated holding periods in days (default: 5,20)",
)
parser.add_argument(
"--exit-price", type=float, help="Manual exit price for single signal recording"
)
parser.add_argument(
"--exit-date", help="Manual exit date (YYYY-MM-DD) for single signal recording"
)
parser.add_argument("--outcome-notes", default="", help="Notes about the outcome")
parser.add_argument(
"--list-ready", action="store_true", help="List signals ready for postmortem"
)
parser.add_argument(
"--min-days", type=int, default=5, help="Minimum days for --list-ready (default: 5)"
)
parser.add_argument("--api-key", help="FMP API key (or set FMP_API_KEY env var)")
parser.add_argument(
"--output-dir",
default="reports/",
help="Output directory for postmortem records (default: reports/)",
)
args = parser.parse_args()
# Get API key
api_key = args.api_key or get_fmp_api_key()
# Parse holding periods
holding_periods = [int(p.strip()) for p in args.holding_periods.split(",")]
# Create output directory
output_dir = Path(args.output_dir)
postmortems_dir = output_dir / "postmortems"
postmortems_dir.mkdir(parents=True, exist_ok=True)
# Handle --list-ready mode
if args.list_ready:
if not args.signals_dir:
print("Error: --signals-dir required with --list-ready", file=sys.stderr)
sys.exit(1)
ready = list_ready_signals(args.signals_dir, args.min_days)
print(f"Found {len(ready)} signals ready for postmortem:")
for sig in ready:
print(
f" {sig.get('signal_id', 'N/A')}: {sig.get('ticker', 'N/A')} on {sig.get('signal_date', 'N/A')}"
)
return
# Handle single signal manual recording
if args.signal_id:
if not args.exit_price or not args.exit_date:
print(
"Error: --exit-price and --exit-date required for manual recording", file=sys.stderr
)
sys.exit(1)
# Create minimal signal record
signal = {
"signal_id": args.signal_id,
"ticker": args.signal_id.split("_")[1] if "_" in args.signal_id else "UNKNOWN",
"signal_date": args.signal_id.split("_")[2]
if args.signal_id.count("_") >= 2
else datetime.now().strftime("%Y-%m-%d"),
"predicted_direction": "LONG",
"source_skill": "manual",
"entry_price": 0.0,
}
postmortem = create_postmortem_record(
signal, {}, args.exit_price, args.exit_date, outcome_notes=args.outcome_notes
)
# Save postmortem
output_file = postmortems_dir / f"{postmortem['postmortem_id']}.json"
with open(output_file, "w", encoding="utf-8") as f:
json.dump(postmortem, f, indent=2)
print(f"Saved postmortem: {output_file}")
print(f"Outcome: {postmortem['outcome_category']}")
return
# Handle batch processing
if not args.signals_file:
print("Error: --signals-file required for batch processing", file=sys.stderr)
sys.exit(1)
signals_path = Path(args.signals_file)
if not signals_path.exists():
print(f"Error: Signals file not found: {args.signals_file}", file=sys.stderr)
sys.exit(1)
with open(signals_path) as f:
data = json.load(f)
signals = data if isinstance(data, list) else data.get("signals", [])
if not signals:
print("No signals found in input file", file=sys.stderr)
sys.exit(1)
if not api_key:
print("Warning: No FMP API key available. Using manual entry prices only.", file=sys.stderr)
# Process signals
results = []
for signal in signals:
postmortem = process_signal(signal, holding_periods, api_key)
if postmortem:
results.append(postmortem)
# Save individual postmortem
output_file = postmortems_dir / f"{postmortem['postmortem_id']}.json"
with open(output_file, "w", encoding="utf-8") as f:
json.dump(postmortem, f, indent=2)
# Save batch summary
timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
summary_file = output_dir / f"postmortem_batch_{timestamp}.json"
summary = {
"schema_version": "1.0",
"generated_at": datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + "Z",
"source_file": str(signals_path),
"total_signals": len(signals),
"processed": len(results),
"postmortems": results,
}
with open(summary_file, "w", encoding="utf-8") as f:
json.dump(summary, f, indent=2)
# Print summary
outcomes = {}
for pm in results:
cat = pm["outcome_category"]
outcomes[cat] = outcomes.get(cat, 0) + 1
print(f"Processed {len(results)}/{len(signals)} signals")
print("Outcome distribution:")
for cat, count in sorted(outcomes.items()):
print(f" {cat}: {count}")
print(f"Saved batch summary: {summary_file}")
if __name__ == "__main__":
main()
scripts/tests/conftest.py
"""
Pytest configuration for signal-postmortem tests.
"""
import sys
from pathlib import Path
# Add scripts directory to path for imports
scripts_dir = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(scripts_dir))
scripts/tests/test_cli_workflow.py
"""Deterministic recorder/analyzer CLI regressions with fictional market data."""
import json
import sys
from datetime import datetime, timezone
import postmortem_analyzer as analyzer
import postmortem_recorder as recorder
import pytest
import yaml
from requests.sessions import Session
NOW = datetime(2026, 10, 4, 12)
class FixedDatetime(datetime):
@classmethod
def now(cls, tz=None):
return NOW if tz is None else NOW.replace(tzinfo=timezone.utc).astimezone(tz)
@classmethod
def utcnow(cls):
raise AssertionError("Use timezone-aware UTC now")
@pytest.fixture(autouse=True)
def offline_clock(monkeypatch):
monkeypatch.setattr(recorder, "datetime", FixedDatetime)
monkeypatch.setattr(analyzer, "datetime", FixedDatetime)
monkeypatch.delenv("FMP_API_KEY", raising=False)
def forbidden(*args, **kwargs):
raise AssertionError("Unexpected network request")
monkeypatch.setattr(Session, "request", forbidden)
def cli(monkeypatch, module, *args):
monkeypatch.setattr(sys, "argv", [module.__file__, *map(str, args)])
return module.main()
def write_json(path, data):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data), encoding="utf-8")
def read_json(path):
return json.loads(path.read_text(encoding="utf-8"))
def signal(index=0, **overrides):
return {
"signal_id": f"sig_{index}",
"ticker": f"FICTION{index}",
"signal_date": "2026-09-01",
"source_skill": "fictional-breakout",
"predicted_direction": "LONG",
"entry_price": 100.0,
"regime": "RISK_ON",
**overrides,
}
def test_recorder_to_analyzer_exports_real_feedback(monkeypatch, tmp_path, capsys):
signals_path = tmp_path / "signals.json"
write_json(signals_path, {"signals": [signal(i) for i in range(8)] + [signal(8, ticker="")]})
original = signals_path.read_bytes()
calls = []
def prices(ticker, start, end, api_key):
assert (start, end, api_key) == ("2026-09-01", "2026-09-26", "fictional-key")
calls.append(ticker)
positive = int(ticker.removeprefix("FICTION")) < 6
# Day 5 is Sunday; the forward lookup must use Monday September 7.
return {
"2026-09-07": 110.0 if positive else 90.0,
"2026-09-21": 120.0 if positive else 80.0,
}
monkeypatch.setattr(recorder, "fetch_price_data", prices)
output = tmp_path / "recorded"
assert (
cli(
monkeypatch,
recorder,
"--signals-file",
signals_path,
"--api-key",
"fictional-key",
"--holding-periods",
"5,20",
"--output-dir",
output,
)
is None
)
captured = capsys.readouterr()
assert "Processed 8/9 signals" in captured.out
assert "TRUE_POSITIVE: 6" in captured.out
assert "FALSE_POSITIVE_SEVERE: 2" in captured.out
assert "missing ticker or date" in captured.err
batch = read_json(output / "postmortem_batch_2026-10-04_120000.json")
assert (batch["total_signals"], batch["processed"]) == (9, 8)
assert batch["generated_at"] == "2026-10-04T12:00:00Z"
assert batch["source_file"] == str(signals_path)
records = batch["postmortems"]
assert len(calls) == 8
for index, record in enumerate(records):
assert read_json(output / "postmortems" / f"pm_sig_{index}.json") == record
assert record["recorded_at"] == "2026-10-04T12:00:00Z"
assert record["exit_date"] == "2026-09-07"
assert record["holding_days"] == 6
assert record["realized_returns"] == pytest.approx(
{"5d": 0.1 if index < 6 else -0.1, "20d": 0.2 if index < 6 else -0.2}
)
assert record["outcome_category"] == (
"TRUE_POSITIVE" if index < 6 else "FALSE_POSITIVE_SEVERE"
)
pm_dir = output / "postmortems"
write_json(pm_dir / "pm_stale.json", {**records[0], "recorded_at": "2020-01-01T00:00:00Z"})
(pm_dir / "pm_broken.json").write_text("{broken", encoding="utf-8")
before = {p.name: p.read_bytes() for p in pm_dir.iterdir()}
analysis = tmp_path / "analysis"
assert (
cli(
monkeypatch,
analyzer,
"--postmortems-dir",
pm_dir,
"--days-back",
30,
"--min-sample-size",
8,
"--generate-weight-feedback",
"--generate-improvement-backlog",
"--summary",
"--output-dir",
analysis,
)
is None
)
captured = capsys.readouterr()
assert "Loaded 8 postmortems" in captured.out
assert "Error loading" in captured.err
feedback = read_json(analysis / "weight_feedback_2026-10-04.json")
assert feedback["generated_at"] == "2026-10-04T12:00:00Z"
assert feedback["min_sample_threshold"] == 8
assert feedback["analysis_period"]["total_postmortems"] == 8
assert feedback["confidence"] == "LOW"
assert len(feedback["skill_adjustments"]) == 1
adjustment = feedback["skill_adjustments"][0]
assert (adjustment["skill"], adjustment["sample_size"]) == ("fictional-breakout", 8)
assert adjustment["suggested_weight"] == 1.1
assert adjustment["accuracy"] == 0.75
assert adjustment["false_positive_rate"] == 0.25
backlog = yaml.safe_load((analysis / "skill_improvement_backlog_2026-10-04.yaml").read_text())
assert [b["issue_type"] for b in backlog] == ["overconfidence", "false_positive_cluster"]
assert [b["priority_score"] for b in backlog] == [300, 200]
assert backlog[0]["evidence"] == {
"severe_fp_rate": 0.25,
"severe_fp_count": 2,
"sample_size": 8,
}
assert backlog[1]["evidence"]["regime_correlation"] == "RISK_ON"
assert all(
b["status"] == "pending" and b["generated_at"] == "2026-10-04T12:00:00Z" for b in backlog
)
summary = (analysis / "postmortem_summary_2026-10-04.md").read_text()
assert "| Overall Accuracy | 75.0% |" in summary
assert "| fictional-breakout | 8 | 75.0% | 25.0% | 5.00% |" in summary
assert "| 2026-09 | 8 | 75.0% |" in summary
assert "| FALSE_POSITIVE_SEVERE | 2 | 25.0% |" in summary
assert signals_path.read_bytes() == original
assert {p.name: p.read_bytes() for p in pm_dir.iterdir()} == before
def test_batch_without_key_records_neutral_and_skips_unusable_signals(
monkeypatch, tmp_path, capsys
):
source = tmp_path / "signals.json"
write_json(source, [signal(), signal(1, entry_price=0), signal(2, signal_date="invalid")])
output = tmp_path / "reports"
cli(monkeypatch, recorder, "--signals-file", source, "--output-dir", output)
batch = read_json(output / "postmortem_batch_2026-10-04_120000.json")
assert (batch["processed"], batch["total_signals"]) == (1, 3)
record = batch["postmortems"][0]
assert record["realized_returns"] == {}
assert record["outcome_category"] == "NEUTRAL"
assert record["exit_price"] == 100.0
assert record["exit_date"] == "2026-09-06"
assert sorted(p.name for p in (output / "postmortems").iterdir()) == ["pm_sig_0.json"]
captured = capsys.readouterr()
assert "No FMP API key" in captured.err
assert "No entry price" in captured.err
assert "Invalid signal date" in captured.err
@pytest.mark.parametrize(
"signal_id,ticker,date",
[("sig_FICTION_2026-09-01", "FICTION", "2026-09-01"), ("manual", "UNKNOWN", "2026-10-04")],
)
def test_manual_recording_preserves_notes_without_inventing_returns(
monkeypatch, tmp_path, capsys, signal_id, ticker, date
):
cli(
monkeypatch,
recorder,
"--signal-id",
signal_id,
"--exit-price",
"105",
"--exit-date",
"2026-10-04",
"--outcome-notes",
"fictional manual observation",
"--output-dir",
tmp_path,
)
record = read_json(tmp_path / "postmortems" / f"pm_{signal_id}.json")
assert (record["ticker"], record["signal_date"]) == (ticker, date)
assert (record["exit_price"], record["entry_price"]) == (105, 0)
assert record["realized_returns"] == {}
assert record["outcome_category"] == "NEUTRAL"
assert record["outcome_notes"] == "fictional manual observation"
assert "Outcome: NEUTRAL" in capsys.readouterr().out
def test_list_ready_uses_fixed_cutoff_and_does_not_record(monkeypatch, tmp_path, capsys):
source = tmp_path / "signals"
write_json(
source / "one.json", [signal(signal_date="2026-09-29"), signal(1, signal_date="2026-09-30")]
)
write_json(source / "two.json", {"signals": [signal(2, signal_date="2026-09-28")]})
(source / "broken.json").write_text("{", encoding="utf-8")
output = tmp_path / "reports"
cli(
monkeypatch,
recorder,
"--signals-dir",
source,
"--list-ready",
"--min-days",
"5",
"--output-dir",
output,
)
captured = capsys.readouterr()
assert "Found 2 signals" in captured.out
assert "sig_0:" in captured.out and "sig_2:" in captured.out
assert "sig_1:" not in captured.out
assert "Error reading" in captured.err
assert not list(output.rglob("*.json"))
@pytest.mark.parametrize(
"args,message",
[
((), "--signals-file required"),
(("--list-ready",), "--signals-dir required"),
(("--signal-id", "manual"), "--exit-price and --exit-date required"),
(("--signals-file", "missing.json"), "Signals file not found"),
(("--signals-file", "empty.json"), "No signals found"),
],
)
def test_recorder_errors_write_no_records(monkeypatch, tmp_path, capsys, args, message):
monkeypatch.chdir(tmp_path)
write_json(tmp_path / "empty.json", {"signals": []})
output = tmp_path / "reports"
with pytest.raises(SystemExit) as error:
cli(monkeypatch, recorder, *args, "--output-dir", output)
assert error.value.code == 1
assert message in capsys.readouterr().err
assert not list(output.rglob("*.json"))
def analyzer_fixture(tmp_path):
pm_dir = tmp_path / "postmortems"
write_json(
pm_dir / "pm_one.json",
{
"source_skill": "fictional-breakout",
"signal_date": "2026-09-01",
"outcome_category": "TRUE_POSITIVE",
"realized_returns": {"5d": 0.1},
"recorded_at": "2026-10-01T00:00:00Z",
},
)
return pm_dir
def test_analyzer_default_metrics_and_group_selection(monkeypatch, tmp_path, capsys):
pm_dir = analyzer_fixture(tmp_path)
output = tmp_path / "analysis"
cli(monkeypatch, analyzer, "--postmortems-dir", pm_dir, "--output-dir", output)
captured = capsys.readouterr()
assert "Samples: 1" in captured.out
assert "Accuracy: 100.0%" in captured.out
assert "FP Rate: 0.0%" in captured.out
assert "Avg Return (5d): 10.00%" in captured.out
assert list(output.iterdir()) == []
cli(
monkeypatch,
analyzer,
"--postmortems-dir",
pm_dir,
"--summary",
"--group-by",
"month",
"--generate-weight-feedback",
"--min-sample-size",
"2",
"--output-dir",
output,
)
summary = (output / "postmortem_summary_2026-10-04.md").read_text()
assert "## By Month" in summary
assert "## By Skill" not in summary
assert "| 2026-09 | 1 | 100.0% |" in summary
assert read_json(output / "weight_feedback_2026-10-04.json")["skill_adjustments"] == []
@pytest.mark.parametrize("stale", [False, True])
def test_analyzer_empty_or_stale_inputs_create_no_output(monkeypatch, tmp_path, capsys, stale):
pm_dir = tmp_path / "postmortems"
if stale:
write_json(pm_dir / "pm_stale.json", {"recorded_at": "2020-01-01T00:00:00Z"})
output = tmp_path / "analysis"
with pytest.raises(SystemExit) as error:
cli(monkeypatch, analyzer, "--postmortems-dir", pm_dir, "--output-dir", output)
assert error.value.code == 1
assert "No postmortems found" in capsys.readouterr().err
assert not output.exists()
scripts/tests/test_postmortem_analyzer.py
"""
Tests for postmortem_analyzer.py
"""
import json
from datetime import datetime, timedelta, timezone
import pytest
from postmortem_analyzer import (
analyze_regime_correlation,
calculate_skill_metrics,
generate_improvement_backlog,
generate_summary,
generate_weight_feedback,
)
@pytest.fixture
def sample_postmortems():
"""Generate sample postmortem records for testing."""
return [
# True positives from vcp-screener
{
"postmortem_id": "pm_1",
"signal_id": "sig_1",
"ticker": "AAPL",
"signal_date": "2026-03-01",
"source_skill": "vcp-screener",
"outcome_category": "TRUE_POSITIVE",
"regime_at_signal": "RISK_ON",
"realized_returns": {"5d": 0.032},
},
{
"postmortem_id": "pm_2",
"signal_id": "sig_2",
"ticker": "MSFT",
"signal_date": "2026-03-02",
"source_skill": "vcp-screener",
"outcome_category": "TRUE_POSITIVE",
"regime_at_signal": "RISK_ON",
"realized_returns": {"5d": 0.025},
},
# False positive from vcp-screener
{
"postmortem_id": "pm_3",
"signal_id": "sig_3",
"ticker": "NVDA",
"signal_date": "2026-03-03",
"source_skill": "vcp-screener",
"outcome_category": "FALSE_POSITIVE",
"regime_at_signal": "RISK_OFF",
"realized_returns": {"5d": -0.018},
},
# True positive from canslim-screener
{
"postmortem_id": "pm_4",
"signal_id": "sig_4",
"ticker": "GOOGL",
"signal_date": "2026-03-01",
"source_skill": "canslim-screener",
"outcome_category": "TRUE_POSITIVE",
"regime_at_signal": "RISK_ON",
"realized_returns": {"5d": 0.041},
},
# Regime mismatch from canslim-screener
{
"postmortem_id": "pm_5",
"signal_id": "sig_5",
"ticker": "TSLA",
"signal_date": "2026-03-02",
"source_skill": "canslim-screener",
"outcome_category": "REGIME_MISMATCH",
"regime_at_signal": "RISK_ON",
"realized_returns": {"5d": -0.045},
},
]
@pytest.fixture
def large_postmortem_set():
"""Generate a larger set for testing threshold behaviors."""
postmortems = []
base_date = datetime(2026, 2, 1)
# 30 signals from vcp-screener: 20 TP, 8 FP, 2 neutral
for i in range(30):
if i < 20:
outcome = "TRUE_POSITIVE"
ret = 0.02 + (i * 0.001)
elif i < 28:
outcome = "FALSE_POSITIVE" if i < 26 else "FALSE_POSITIVE_SEVERE"
ret = -0.015 - ((i - 20) * 0.005)
else:
outcome = "NEUTRAL"
ret = 0.002
postmortems.append(
{
"postmortem_id": f"pm_vcp_{i}",
"signal_id": f"sig_vcp_{i}",
"ticker": f"TICK{i}",
"signal_date": (base_date + timedelta(days=i)).strftime("%Y-%m-%d"),
"source_skill": "vcp-screener",
"outcome_category": outcome,
"regime_at_signal": "RISK_OFF" if i >= 20 and i < 28 else "RISK_ON",
"realized_returns": {"5d": ret},
"recorded_at": datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + "Z",
}
)
# 25 signals from canslim-screener: 18 TP, 5 FP, 2 regime mismatch
for i in range(25):
if i < 18:
outcome = "TRUE_POSITIVE"
ret = 0.025 + (i * 0.001)
elif i < 23:
outcome = "FALSE_POSITIVE"
ret = -0.012 - ((i - 18) * 0.003)
else:
outcome = "REGIME_MISMATCH"
ret = -0.035
postmortems.append(
{
"postmortem_id": f"pm_canslim_{i}",
"signal_id": f"sig_canslim_{i}",
"ticker": f"CAN{i}",
"signal_date": (base_date + timedelta(days=i)).strftime("%Y-%m-%d"),
"source_skill": "canslim-screener",
"outcome_category": outcome,
"regime_at_signal": "RISK_ON",
"realized_returns": {"5d": ret},
"recorded_at": datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + "Z",
}
)
return postmortems
class TestCalculateSkillMetrics:
"""Tests for calculate_skill_metrics function."""
def test_basic_metrics(self, sample_postmortems):
"""Test basic metric calculation."""
metrics = calculate_skill_metrics(sample_postmortems)
assert "vcp-screener" in metrics
assert "canslim-screener" in metrics
vcp = metrics["vcp-screener"]
assert vcp["sample_size"] == 3
assert vcp["true_positive"] == 2
assert vcp["false_positive"] == 1
assert vcp["accuracy"] == pytest.approx(2 / 3, rel=1e-6)
canslim = metrics["canslim-screener"]
assert canslim["sample_size"] == 2
assert canslim["true_positive"] == 1
assert canslim["regime_mismatch"] == 1
def test_average_return_calculation(self, sample_postmortems):
"""Test average return calculation."""
metrics = calculate_skill_metrics(sample_postmortems)
vcp = metrics["vcp-screener"]
expected_avg = (0.032 + 0.025 - 0.018) / 3
assert vcp["avg_return_5d"] == pytest.approx(expected_avg, rel=1e-6)
class TestGenerateWeightFeedback:
"""Tests for generate_weight_feedback function."""
def test_no_feedback_below_threshold(self, sample_postmortems):
"""Test that no adjustments are made with small sample size."""
metrics = calculate_skill_metrics(sample_postmortems)
feedback = generate_weight_feedback(metrics, min_sample_size=20)
assert feedback["skill_adjustments"] == []
assert feedback["confidence"] == "LOW"
def test_feedback_with_sufficient_samples(self, large_postmortem_set):
"""Test weight feedback generation with sufficient samples."""
metrics = calculate_skill_metrics(large_postmortem_set)
feedback = generate_weight_feedback(metrics, min_sample_size=20)
assert len(feedback["skill_adjustments"]) > 0
assert feedback["confidence"] in ("MEDIUM", "HIGH")
# Check structure
for adj in feedback["skill_adjustments"]:
assert "skill" in adj
assert "current_weight" in adj
assert "suggested_weight" in adj
assert "reason" in adj
assert 0.3 <= adj["suggested_weight"] <= 2.0
def test_schema_version(self, large_postmortem_set):
"""Test that output includes schema version."""
metrics = calculate_skill_metrics(large_postmortem_set)
feedback = generate_weight_feedback(metrics)
assert feedback["schema_version"] == "1.0"
assert "generated_at" in feedback
class TestGenerateImprovementBacklog:
"""Tests for generate_improvement_backlog function."""
def test_backlog_with_high_fp_rate(self, large_postmortem_set):
"""Test backlog generation for high false positive rate."""
metrics = calculate_skill_metrics(large_postmortem_set)
backlog = generate_improvement_backlog(metrics, large_postmortem_set, min_sample_size=15)
# Should have at least one entry for vcp-screener FP cluster
assert len(backlog) > 0
# Check structure
for entry in backlog:
assert "skill" in entry
assert "issue_type" in entry
assert "severity" in entry
assert "evidence" in entry
assert "suggested_action" in entry
assert "priority_score" in entry
assert "generated_by" in entry
assert entry["generated_by"] == "signal-postmortem"
def test_backlog_sorted_by_priority(self, large_postmortem_set):
"""Test that backlog is sorted by priority score descending."""
metrics = calculate_skill_metrics(large_postmortem_set)
backlog = generate_improvement_backlog(metrics, large_postmortem_set, min_sample_size=15)
if len(backlog) > 1:
for i in range(len(backlog) - 1):
assert backlog[i]["priority_score"] >= backlog[i + 1]["priority_score"]
class TestAnalyzeRegimeCorrelation:
"""Tests for analyze_regime_correlation function."""
def test_finds_risk_off_correlation(self, large_postmortem_set):
"""Test that RISK_OFF correlation is detected for vcp-screener."""
result = analyze_regime_correlation(large_postmortem_set, "vcp-screener")
# VCP-screener has FPs concentrated in RISK_OFF regime
assert result == "RISK_OFF"
def test_no_correlation_for_clean_skill(self, sample_postmortems):
"""Test no correlation for skill with too few samples per regime."""
result = analyze_regime_correlation(sample_postmortems, "canslim-screener")
# Not enough samples to establish correlation
assert result == "NONE"
class TestGenerateSummary:
"""Tests for generate_summary function."""
def test_summary_contains_required_sections(self, large_postmortem_set):
"""Test that summary contains all required sections."""
metrics = calculate_skill_metrics(large_postmortem_set)
summary = generate_summary(metrics, large_postmortem_set, ["skill", "month"])
assert "# Signal Postmortem Summary" in summary
assert "## Overall Statistics" in summary
assert "## By Skill" in summary
assert "## By Month" in summary
assert "## Outcome Distribution" in summary
def test_summary_markdown_table_format(self, large_postmortem_set):
"""Test that summary uses proper markdown table format."""
metrics = calculate_skill_metrics(large_postmortem_set)
summary = generate_summary(metrics, large_postmortem_set, ["skill"])
# Check for table headers
assert "| Skill | Samples |" in summary
assert "|-------|---------|" in summary
class TestIntegration:
"""Integration tests for analyzer flow."""
def test_full_analysis_flow(self, tmp_path, large_postmortem_set):
"""Test complete analysis flow with file output."""
# Write postmortems to temp directory
pm_dir = tmp_path / "postmortems"
pm_dir.mkdir()
for pm in large_postmortem_set:
with open(pm_dir / f"{pm['postmortem_id']}.json", "w", encoding="utf-8") as f:
json.dump(pm, f)
# Calculate metrics
metrics = calculate_skill_metrics(large_postmortem_set)
# Generate all outputs
feedback = generate_weight_feedback(metrics, min_sample_size=20)
backlog = generate_improvement_backlog(metrics, large_postmortem_set, min_sample_size=15)
summary = generate_summary(metrics, large_postmortem_set, ["skill", "month"])
# Verify outputs are non-empty and well-formed
assert len(feedback["skill_adjustments"]) >= 0 # May or may not have adjustments
assert len(backlog) >= 0 # May or may not have issues
assert len(summary) > 100 # Should have substantial content
scripts/tests/test_postmortem_recorder.py
"""
Tests for postmortem_recorder.py
"""
import json
import pytest
from postmortem_recorder import calculate_return, classify_outcome, create_postmortem_record
class TestCalculateReturn:
"""Tests for calculate_return function."""
def test_positive_return(self):
"""Test positive return calculation."""
result = calculate_return(100.0, 110.0)
assert result == pytest.approx(0.10, rel=1e-6)
def test_negative_return(self):
"""Test negative return calculation."""
result = calculate_return(100.0, 95.0)
assert result == pytest.approx(-0.05, rel=1e-6)
def test_zero_entry_price(self):
"""Test handling of zero entry price."""
result = calculate_return(0.0, 100.0)
assert result == 0.0
def test_no_change(self):
"""Test flat return."""
result = calculate_return(100.0, 100.0)
assert result == 0.0
class TestClassifyOutcome:
"""Tests for classify_outcome function."""
def test_true_positive_long(self):
"""Test TRUE_POSITIVE classification for LONG signal with positive return."""
result = classify_outcome(
predicted_direction="LONG",
return_pct=0.05,
regime_at_signal="RISK_ON",
regime_at_exit="RISK_ON",
)
assert result == "TRUE_POSITIVE"
def test_true_positive_short(self):
"""Test TRUE_POSITIVE classification for SHORT signal with negative return."""
result = classify_outcome(
predicted_direction="SHORT",
return_pct=-0.03,
regime_at_signal="RISK_OFF",
regime_at_exit="RISK_OFF",
)
assert result == "TRUE_POSITIVE"
def test_false_positive_long(self):
"""Test FALSE_POSITIVE classification for LONG signal with negative return."""
result = classify_outcome(
predicted_direction="LONG",
return_pct=-0.015,
regime_at_signal="RISK_ON",
regime_at_exit="RISK_ON",
)
assert result == "FALSE_POSITIVE"
def test_false_positive_severe(self):
"""Test FALSE_POSITIVE_SEVERE classification for large loss."""
result = classify_outcome(
predicted_direction="LONG",
return_pct=-0.05,
regime_at_signal="RISK_ON",
regime_at_exit="RISK_ON",
)
assert result == "FALSE_POSITIVE_SEVERE"
def test_neutral_flat_return(self):
"""Test NEUTRAL classification for flat return."""
result = classify_outcome(
predicted_direction="LONG",
return_pct=0.002,
regime_at_signal="RISK_ON",
regime_at_exit="RISK_ON",
)
assert result == "NEUTRAL"
def test_regime_mismatch(self):
"""Test REGIME_MISMATCH classification when regime changed."""
result = classify_outcome(
predicted_direction="LONG",
return_pct=-0.03,
regime_at_signal="RISK_ON",
regime_at_exit="RISK_OFF",
)
assert result == "REGIME_MISMATCH"
def test_regime_change_but_small_loss(self):
"""Test that small loss with regime change is not REGIME_MISMATCH."""
result = classify_outcome(
predicted_direction="LONG",
return_pct=-0.01,
regime_at_signal="RISK_ON",
regime_at_exit="RISK_OFF",
)
# Should be FALSE_POSITIVE since loss is < 2%
assert result == "FALSE_POSITIVE"
class TestCreatePostmortemRecord:
"""Tests for create_postmortem_record function."""
def test_basic_postmortem_creation(self):
"""Test basic postmortem record creation."""
signal = {
"signal_id": "sig_aapl_20260310_abc",
"ticker": "AAPL",
"signal_date": "2026-03-10",
"predicted_direction": "LONG",
"source_skill": "vcp-screener",
"entry_price": 170.0,
"regime": "RISK_ON",
}
realized_returns = {"5d": 0.032, "20d": 0.058}
result = create_postmortem_record(
signal=signal,
realized_returns=realized_returns,
exit_price=175.44,
exit_date="2026-03-15",
)
assert result["schema_version"] == "1.0"
assert result["postmortem_id"] == "pm_sig_aapl_20260310_abc"
assert result["signal_id"] == "sig_aapl_20260310_abc"
assert result["ticker"] == "AAPL"
assert result["signal_date"] == "2026-03-10"
assert result["source_skill"] == "vcp-screener"
assert result["predicted_direction"] == "LONG"
assert result["entry_price"] == 170.0
assert result["realized_returns"]["5d"] == 0.032
assert result["exit_price"] == 175.44
assert result["exit_date"] == "2026-03-15"
assert result["holding_days"] == 5
assert result["outcome_category"] == "TRUE_POSITIVE"
assert result["regime_at_signal"] == "RISK_ON"
assert "recorded_at" in result
def test_postmortem_with_false_positive(self):
"""Test postmortem record with false positive outcome."""
signal = {
"signal_id": "sig_nvda_20260305_xyz",
"ticker": "NVDA",
"signal_date": "2026-03-05",
"predicted_direction": "LONG",
"source_skill": "canslim-screener",
"entry_price": 900.0,
"regime": "RISK_ON",
}
realized_returns = {"5d": -0.033}
result = create_postmortem_record(
signal=signal,
realized_returns=realized_returns,
exit_price=870.3,
exit_date="2026-03-10",
)
assert result["outcome_category"] == "FALSE_POSITIVE_SEVERE"
assert result["holding_days"] == 5
def test_postmortem_missing_dates(self):
"""Test postmortem with missing dates."""
signal = {
"signal_id": "sig_test_abc",
"ticker": "TEST",
"signal_date": "",
"predicted_direction": "LONG",
"source_skill": "test",
"entry_price": 100.0,
}
result = create_postmortem_record(
signal=signal, realized_returns={}, exit_price=105.0, exit_date=""
)
assert result["holding_days"] == 0
assert result["outcome_category"] == "NEUTRAL" # No returns data
class TestIntegration:
"""Integration tests for postmortem recording flow."""
def test_full_postmortem_flow(self, tmp_path):
"""Test complete postmortem recording to file."""
signal = {
"signal_id": "sig_msft_20260301_test",
"ticker": "MSFT",
"signal_date": "2026-03-01",
"predicted_direction": "LONG",
"source_skill": "earnings-trade-analyzer",
"entry_price": 420.0,
"regime": "RISK_ON",
}
realized_returns = {"5d": 0.024, "20d": 0.045}
postmortem = create_postmortem_record(
signal=signal,
realized_returns=realized_returns,
exit_price=430.08,
exit_date="2026-03-06",
)
# Write to file
output_file = tmp_path / "pm_sig_msft_20260301_test.json"
with open(output_file, "w", encoding="utf-8") as f:
json.dump(postmortem, f, indent=2)
# Read back and verify
with open(output_file) as f:
loaded = json.load(f)
assert loaded["ticker"] == "MSFT"
assert loaded["outcome_category"] == "TRUE_POSITIVE"
assert loaded["realized_returns"]["5d"] == 0.024
SKILL.md
---
name: signal-postmortem
description: Record and analyze post-trade outcomes for signals generated by edge pipeline and other skills. Track false positives, missed opportunities, and regime mismatches. Feed results back to edge-signal-aggregator weights and skill improvement backlog.
---
# Signal Postmortem
## Overview
Signal Postmortem records and analyzes the outcomes of trading signals generated by the edge pipeline, screeners, and other skills. It compares predicted edge direction against 5-day and 20-day realized returns, categorizes outcomes (true positive, false positive, missed opportunity, regime mismatch), and generates feedback for edge-signal-aggregator weight adjustments and skill improvement backlog entries.
## When to Use
- After a trade has been closed and you want to record the outcome
- When reviewing a batch of signals that have reached their holding period (5 or 20 days)
- To identify systematic false positive patterns from specific skills
- To generate feedback for edge-signal-aggregator weight calibration
- When building a skill improvement backlog from decision quality metrics
- For periodic (weekly/monthly) signal quality audits
## Prerequisites
- Python 3.9+
- FMP API key (optional, for fetching realized returns if not provided manually)
- Standard library + `requests` for API calls
- Input: signal records in JSON format (from edge-signal-aggregator or screener outputs)
### API Key Setup (Optional)
If you want to automatically fetch price data for return calculations, set up the FMP API key:
```bash
export FMP_API_KEY=your_api_key_here
```
Alternatively, pass the key via command line with `--api-key YOUR_KEY`. Without an API key, you can still record outcomes manually by providing `--exit-price` and `--exit-date`.
## Workflow
### Step 1: Prepare Signal Records
Gather closed or matured signal records. Each record should include:
- `signal_id`: Unique identifier
- `ticker`: Stock symbol
- `signal_date`: Date signal was generated
- `predicted_direction`: LONG or SHORT
- `source_skill`: Which skill generated the signal
- `entry_price`: Price at signal generation (optional, for manual override)
```bash
# Example: List signals ready for postmortem (5+ days old)
python3 skills/signal-postmortem/scripts/postmortem_recorder.py \
--list-ready \
--signals-dir state/signals/ \
--min-days 5
```
### Step 2: Record Outcomes
Run the postmortem recorder to fetch realized returns and classify outcomes.
```bash
python3 skills/signal-postmortem/scripts/postmortem_recorder.py \
--signals-file state/signals/aggregated_signals_2026-03-10.json \
--holding-periods 5,20 \
--output-dir reports/
```
For manual outcome recording (when price data is already available):
```bash
python3 skills/signal-postmortem/scripts/postmortem_recorder.py \
--signal-id sig_aapl_20260310_abc \
--exit-price 178.50 \
--exit-date 2026-03-15 \
--outcome-notes "Closed at target, +3.2% in 5 days" \
--output-dir reports/
```
### Step 3: Classify Outcomes
The recorder automatically classifies each signal into one of four categories:
| Category | Definition |
|----------|------------|
| TRUE_POSITIVE | Predicted direction matched realized return sign |
| FALSE_POSITIVE | Predicted direction opposite to realized return |
| MISSED_OPPORTUNITY | Signal not taken but would have been profitable |
| REGIME_MISMATCH | Signal failed due to market regime change |
Classification rules are documented in `references/outcome-classification.md`.
### Step 4: Generate Feedback Files
Generate feedback for downstream consumers:
```bash
# Generate weight adjustment suggestions for edge-signal-aggregator
python3 skills/signal-postmortem/scripts/postmortem_analyzer.py \
--postmortems-dir reports/postmortems/ \
--generate-weight-feedback \
--output-dir reports/
# Generate skill improvement backlog entries
python3 skills/signal-postmortem/scripts/postmortem_analyzer.py \
--postmortems-dir reports/postmortems/ \
--generate-improvement-backlog \
--output-dir reports/
```
### Step 5: Review Summary Statistics
Generate aggregate statistics by skill, by ticker, and by time period:
```bash
python3 skills/signal-postmortem/scripts/postmortem_analyzer.py \
--postmortems-dir reports/postmortems/ \
--summary \
--group-by skill,month \
--output-dir reports/
```
## Output Format
### Postmortem Record (JSON)
```json
{
"schema_version": "1.0",
"postmortem_id": "pm_sig_aapl_20260310_abc",
"signal_id": "sig_aapl_20260310_abc",
"ticker": "AAPL",
"signal_date": "2026-03-10",
"source_skill": "edge-signal-aggregator",
"predicted_direction": "LONG",
"entry_price": 172.50,
"realized_returns": {
"5d": 0.032,
"20d": 0.058
},
"exit_price": 178.50,
"exit_date": "2026-03-15",
"holding_days": 5,
"outcome_category": "TRUE_POSITIVE",
"regime_at_signal": "RISK_ON",
"regime_at_exit": "RISK_ON",
"outcome_notes": "Clean breakout, held through minor pullback",
"recorded_at": "2026-03-17T10:30:00Z"
}
```
### Weight Feedback (JSON)
```json
{
"schema_version": "1.0",
"generated_at": "2026-03-17T10:35:00Z",
"analysis_period": {
"from": "2026-02-01",
"to": "2026-03-15"
},
"skill_adjustments": [
{
"skill": "vcp-screener",
"current_weight": 1.0,
"suggested_weight": 0.85,
"reason": "15% false positive rate in RISK_OFF regime",
"sample_size": 42
}
],
"confidence": "MEDIUM",
"min_sample_threshold": 20
}
```
### Skill Improvement Backlog Entry (YAML)
```yaml
- skill: vcp-screener
issue_type: false_positive_cluster
severity: medium
evidence:
false_positive_rate: 0.15
sample_size: 42
regime_correlation: RISK_OFF
suggested_action: "Add regime filter or reduce signal confidence in RISK_OFF"
generated_by: signal-postmortem
generated_at: "2026-03-17T10:35:00Z"
```
### Summary Report (Markdown)
Reports are saved to `reports/` with filenames `postmortem_summary_YYYY-MM-DD.md`.
## Resources
- `scripts/postmortem_recorder.py` -- Records individual signal outcomes
- `scripts/postmortem_analyzer.py` -- Generates feedback and summary statistics
- `references/outcome-classification.md` -- Classification rules and edge cases
- `references/feedback-integration.md` -- How to integrate feedback with downstream skills
## Key Principles
1. **Honest Attribution** -- Every outcome is attributed to its source skill for accountability
2. **Regime Awareness** -- Regime context is recorded to distinguish skill failure from market regime shifts
3. **Minimum Sample Size** -- Weight adjustments require 20+ signals for statistical validity
4. **Feedback Loop Closure** -- Results flow back to improve both signal aggregation and skill quality