assets/performance_coach_report.schema.json
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "Trade Performance Coach Report",
"type": "object",
"required": [
"schema_version",
"review_type",
"review_id",
"overall_verdict",
"summary",
"scores",
"process_adherence_findings",
"risk_manager_notes",
"execution_quality_assessment",
"behavioral_pattern_tags",
"next_session_operating_rules",
"coach_questions",
"human_decision_gate",
"disclaimer"
],
"properties": {
"schema_version": {"const": 1},
"review_type": {"enum": ["single_trade", "partial_close", "monthly_aggregate"]},
"review_id": {"type": "string", "minLength": 1},
"generated_at": {"type": "string"},
"source_records": {"type": "array", "items": {"type": "string"}},
"overall_verdict": {"enum": ["OK", "WARN", "REVIEW_REQUIRED", "RULE_VIOLATION", "COOL_DOWN"]},
"summary": {
"type": "object",
"required": ["outcome", "primary_root_cause", "secondary_root_causes", "confidence"],
"properties": {
"outcome": {"enum": ["win", "loss", "breakeven", "open", "mixed", "unknown"]},
"primary_root_cause": {"enum": ["thesis_quality", "execution", "risk_sizing", "market_environment", "rule_violation", "randomness", "unknown"]},
"secondary_root_causes": {"type": "array", "items": {"type": "string"}},
"confidence": {"enum": ["low", "medium", "high"]}
},
"additionalProperties": true
},
"scores": {
"type": "object",
"required": ["process_score", "risk_score", "execution_score", "review_quality_score"],
"properties": {
"process_score": {"type": "integer", "minimum": 0, "maximum": 100},
"risk_score": {"type": "integer", "minimum": 0, "maximum": 100},
"execution_score": {"type": "integer", "minimum": 0, "maximum": 100},
"review_quality_score": {"type": "integer", "minimum": 0, "maximum": 100}
},
"additionalProperties": false
},
"process_adherence_findings": {"type": "array", "items": {"$ref": "#/$defs/finding"}},
"risk_manager_notes": {"type": "array", "items": {"$ref": "#/$defs/riskNote"}},
"execution_quality_assessment": {"type": "array", "items": {"$ref": "#/$defs/executionFinding"}},
"behavioral_pattern_tags": {"type": "array", "items": {"$ref": "#/$defs/behaviorTag"}},
"next_session_operating_rules": {"type": "array", "items": {"$ref": "#/$defs/operatingRule"}},
"coach_questions": {"type": "array", "items": {"type": "string"}},
"human_decision_gate": {
"type": "object",
"required": ["question", "allowed_actions", "default_action"],
"properties": {
"question": {"type": "string"},
"allowed_actions": {
"type": "array",
"items": {"enum": ["accept_rules", "modify_rules", "defer", "journal_only"]}
},
"default_action": {"const": "journal_only"}
},
"additionalProperties": false
},
"disclaimer": {"type": "string"}
},
"$defs": {
"finding": {
"type": "object",
"required": ["rule", "status", "evidence", "severity"],
"properties": {
"rule": {"type": "string"},
"status": {"enum": ["met", "missed", "unclear", "not_applicable"]},
"evidence": {"type": "string"},
"severity": {"enum": ["info", "warning", "critical"]}
},
"additionalProperties": false
},
"riskNote": {
"type": "object",
"required": ["topic", "finding", "severity", "evidence"],
"properties": {
"topic": {"enum": ["position_size", "portfolio_heat", "drawdown", "correlation", "stop_discipline", "loss_limit", "regime_gate"]},
"finding": {"type": "string"},
"severity": {"enum": ["info", "warning", "critical"]},
"evidence": {"type": "string"}
},
"additionalProperties": false
},
"executionFinding": {
"type": "object",
"required": ["phase", "finding", "evidence", "severity"],
"properties": {
"phase": {"enum": ["entry", "add", "trim", "stop", "exit", "review"]},
"finding": {"type": "string"},
"evidence": {"type": "string"},
"severity": {"enum": ["info", "warning", "critical"]}
},
"additionalProperties": false
},
"behaviorTag": {
"type": "object",
"required": ["tag", "confidence", "evidence", "reflection_question"],
"properties": {
"tag": {"enum": ["fomo_entry", "revenge_trade", "premature_exit", "overconfidence_after_winner", "loss_aversion", "rule_drift", "stop_moved", "size_creep", "unknown_size_discipline", "hesitation", "no_pattern_detected"]},
"confidence": {"enum": ["low", "medium", "high"]},
"evidence": {"type": "string"},
"reflection_question": {"type": "string"}
},
"additionalProperties": false
},
"operatingRule": {
"type": "object",
"required": ["rule", "duration", "trigger", "reason"],
"properties": {
"rule": {"type": "string"},
"duration": {"type": "string"},
"trigger": {"type": "string"},
"reason": {"type": "string"}
},
"additionalProperties": false
}
},
"additionalProperties": true
}
references/behavior-tags.md
# Trading Behavior Tags
## Principle
Behavior tags are not psychological diagnoses. They are evidence-based hypotheses
about trading behavior that may explain process drift. Always use language like
"possible pattern" and include evidence plus a reflection question.
## MVP Tags
| Tag | Meaning | Evidence Examples | Guardrail Ideas |
|---|---|---|---|
| `fomo_entry` | Entered because of fear of missing a move rather than setup confirmation. | journal says "didn't want to miss", entry before confirmation, chased above plan | require thesis + screenshot before entry; no chase after missed trigger |
| `revenge_trade` | Trade appears motivated by making back a prior loss. | journal says "make it back", trade immediately after loss, oversized after loss | review-only mode after two losses; 24h delay after max loss |
| `premature_exit` | Exited before plan due to discomfort without invalidation. | exit before stop/target; journal says "got scared" | predefine partial/exit rules; no discretionary exit without invalidation note |
| `overconfidence_after_winner` | Risk increased after a win without rule justification. | actual risk > plan after prior winner; journal says "easy money" | cap risk after large winner; require normal sizing until next review |
| `stop_moved` | Stop was moved after entry without a pre-defined plan. | `actual.stop_moved = true` and not planned | stop changes must be pre-written; unplanned move triggers review |
| `size_creep` | Actual risk or size exceeded risk plan. | actual R > planned max R (explicit comparison) | reduce next trade size; require position-size screenshot |
| `unknown_size_discipline` | Risk discipline could not be assessed because actual.risk_r or the risk plan reference is missing. | position_size warning emitted because actual or reference risk is None | record planned and actual risk on the next trade so size discipline becomes verifiable |
| `hesitation` | Valid plan was not executed due to hesitation. | missed planned entry; journal says "hesitated" | use if/then trigger checklist; define no-entry after missed trigger |
| `rule_drift` | Repeated small deviations from rules. | multiple minor deviations or unclear records | simplify rules; pick one rule to enforce next week |
| `no_pattern_detected` | No behavior pattern detected from available evidence. | clean record with no supportive evidence | no behavior-specific guardrail needed |
## Tagging Rules
1. Never infer personality traits.
2. Never claim a medical or mental-health diagnosis.
3. Use low/medium/high confidence based on evidence strength.
4. Include exactly what evidence supports each tag.
5. Include a reflection question.
6. If evidence is weak, use `confidence: low` or do not tag.
## Reflection Question Examples
- FOMO: "What evidence did you have before entry that would not have been available after waiting for confirmation?"
- Revenge: "Would you have taken this trade if the prior trade had been a winner?"
- Premature exit: "What rule did the exit satisfy, or was it driven by discomfort?"
- Overconfidence: "Did the prior win change the size or quality threshold for this trade?"
- Stop moved: "Was the new stop part of the plan before entry?"
- Size creep: "What would this trade have looked like at the planned risk size?"
references/hermes-integration.md
# Hermes Integration Notes
## Suggested Slash Command: `/post-trade-coach`
Purpose: review a closed or partially closed trade and generate process, risk,
execution, and possible behavior-pattern feedback.
Suggested bundle flow:
```text
trader-memory-core
→ signal-postmortem
→ trade-performance-coach
```
Required output sections:
- Data freshness / source provenance
- Trade summary
- Process adherence
- Risk manager notes
- Execution quality
- Possible behavioral patterns
- Coach questions
- Next-session operating rules
- Human decision gate
## Suggested Slash Command: `/monthly-performance-coach`
Purpose: review recurring process, risk, and behavior patterns across the month.
Suggested bundle flow:
```text
trader-memory-core
→ signal-postmortem
→ trade-performance-coach
→ monthly-performance-review
```
## Guardrails for Hermes Prompts
Hermes prompts should state:
- This is not financial advice.
- This is not therapy or mental-health diagnosis.
- The assistant must not place or prepare broker orders.
- The assistant must not tell the user to buy, sell, short, hold, or size a specific security.
- All recommendations are process guardrails for the human trader to accept, modify, defer, or journal.
## Naming
Existing alias can remain:
```text
trading-research-assistant
```
Future public positioning can broaden to:
```text
Hermes Trading Desk Assistant
```
Suggested subtitle:
```text
A trading desk assistant for research, journaling, risk review, and performance coaching.
```
references/output-contract.md
# Output Contract
## Top-Level JSON
```json
{
"schema_version": 1,
"review_type": "single_trade | partial_close | monthly_aggregate",
"review_id": "string",
"source_records": ["string"],
"overall_verdict": "OK | WARN | REVIEW_REQUIRED | RULE_VIOLATION | COOL_DOWN",
"summary": {
"outcome": "win | loss | breakeven | open | mixed | unknown",
"primary_root_cause": "thesis_quality | execution | risk_sizing | market_environment | rule_violation | randomness | unknown",
"secondary_root_causes": ["string"],
"confidence": "low | medium | high"
},
"scores": {
"process_score": 0,
"risk_score": 0,
"execution_score": 0,
"review_quality_score": 0
},
"process_adherence_findings": [],
"risk_manager_notes": [],
"execution_quality_assessment": [],
"behavioral_pattern_tags": [],
"next_session_operating_rules": [],
"coach_questions": [],
"human_decision_gate": {
"question": "string",
"allowed_actions": ["accept_rules", "modify_rules", "defer", "journal_only"],
"default_action": "journal_only"
},
"disclaimer": "string"
}
```
## Findings
Each finding should include:
```json
{
"rule": "string",
"status": "met | missed | unclear | not_applicable",
"evidence": "string",
"severity": "info | warning | critical"
}
```
Risk notes use:
```json
{
"topic": "position_size | portfolio_heat | drawdown | correlation | stop_discipline | loss_limit | regime_gate",
"finding": "string",
"severity": "info | warning | critical",
"evidence": "string"
}
```
Behavior tags use:
```json
{
"tag": "fomo_entry | revenge_trade | premature_exit | overconfidence_after_winner | loss_aversion | rule_drift | stop_moved | size_creep | unknown_size_discipline | hesitation | no_pattern_detected",
"confidence": "low | medium | high",
"evidence": "string",
"reflection_question": "string"
}
```
## Markdown Report
Markdown reports should include:
1. Verdict
2. Trade / period summary
3. Process adherence
4. Risk manager notes
5. Execution quality
6. Possible behavior patterns
7. Next-session operating rules
8. Coach questions
9. Human decision gate
10. Disclaimer
references/review-framework.md
# Trade Performance Coach Review Framework
## Purpose
This framework turns recorded trade evidence into process-improvement feedback.
It evaluates the quality of the trader's decision process, not whether the trade
made money.
A clean-process loss can be acceptable. A profitable rule violation can still be
a serious process problem.
## Five-Axis Review Model
| Axis | Question | Typical Evidence | Output |
|---|---|---|---|
| Thesis Quality | Was the original idea clear, falsifiable, and aligned with the setup? | trade thesis, invalidation, catalyst, setup notes | root-cause finding |
| Process Adherence | Did actual behavior follow the plan and workflow? | entry timing, setup confirmation, thesis record, stop plan | process findings |
| Risk Discipline | Did actual risk follow predefined limits? | planned R, actual R, portfolio heat, weekly loss, correlations | risk manager notes |
| Execution Quality | Were entry, stop, adds/trims, and exits consistent with plan? | execution log, partial close, stop changes | execution assessment |
| Behavior Pattern | Did the journal suggest repeated behavior patterns? | journal text, timing, rule deviations | behavior tags + questions |
## Root-Cause Categories
Use one primary root cause and optional secondary causes:
- `thesis_quality`
- `execution`
- `risk_sizing`
- `market_environment`
- `rule_violation`
- `randomness`
- `unknown`
Classification guidance:
- If the plan was followed and the setup was valid, but the market simply moved
against the trade, prefer `randomness` or `market_environment`.
- If the trade was taken without the required setup confirmation, prefer
`execution` or `rule_violation`.
- If the actual risk exceeded the user-defined plan, prefer `risk_sizing`.
- If the user's note shows the trade was taken to make back a prior loss, prefer
`rule_violation` with a possible `revenge_trade` behavior tag.
## Verdicts
| Verdict | Use When |
|---|---|
| `OK` | The record is complete enough, no material process/risk issue appears, and no behavior pattern is detected. |
| `WARN` | Minor issue, incomplete evidence, or soft process concern. |
| `REVIEW_REQUIRED` | Meaningful process/risk/execution issue, but not a hard rule violation. |
| `RULE_VIOLATION` | Explicit rule breach: risk limit, stop discipline, regime gate, or unplanned add/trim. |
| `COOL_DOWN` | Repeated violations, revenge pattern, drawdown escalation, or multiple critical findings. |
## Scoring
Scores are advisory, not scientific. They help the user compare reviews over
time.
Start at 100 and subtract:
- info finding: 0-5
- warning finding: 10-20
- critical finding: 25-40
- missing critical record: 10-20
Recommended score fields:
```yaml
process_score: 0-100
risk_score: 0-100
execution_score: 0-100
review_quality_score: 0-100
```
`review_quality_score` reflects evidence completeness. Do not over-interpret
behavior tags when review quality is low.
## Next-Session Operating Rules
Each review should produce concrete, temporary operating rules. Good rules are:
- observable
- time-boxed
- tied to evidence
- simple enough to follow before the next trade
Examples:
- "For the next two trades, max planned risk is 0.5R."
- "No entry without a pre-entry thesis record and invalidation point."
- "If two rule violations occur in one week, switch to review-only mode."
- "A missed entry may be added to the watchlist, but not chased above the planned trigger."
## Evidence Standard
Every warning or tag should cite evidence:
- field value (`actual.risk_r = 1.8` vs `max_risk_per_trade_r = 1.0`)
- journal phrase
- missing required record
- explicit flag (`stop_moved = true`)
Avoid unsupported claims. If evidence is unclear, say so.
references/risk-review-checklist.md
# Risk Review Checklist
## Purpose
This checklist gives the skill a risk-manager style review framework. It compares
actual actions with the user's own documented rules. It does not create financial
advice or decide whether a trade should be taken.
## Checks
| Check | Required Evidence | Warning | Critical |
|---|---|---|---|
| Per-trade risk | planned max R and actual R | actual > planned by <=25% | actual > planned by >25% |
| Portfolio heat | max heat and actual heat | actual near max | actual > max |
| Weekly loss limit | max weekly loss and current weekly loss | within 10% of limit | breached limit |
| Consecutive losses | count and threshold | near threshold | at/above threshold |
| Regime gate | market regime and allowed trade types | unclear gate | trade taken against restrictive/cash-priority gate |
| Stop discipline | planned stop and actual stop changes | unclear stop change | unplanned stop move |
| Adds/trims | original add/trim rules | discretionary add/trim unclear | add to loser or add without plan |
| Correlation cluster | holdings and sector/theme exposure | concentrated exposure | new trade increases high-risk cluster |
## Severity Guidance
`info`: observation only. No rule concern.
`warning`: possible issue or incomplete evidence. Human should review.
`critical`: explicit user rule violation or repeated pattern. Should produce
`RULE_VIOLATION` or `COOL_DOWN` depending on recurrence.
## Risk Manager Language
Use objective rule language:
- Good: "Actual risk was 1.8R versus a stated max of 1.0R."
- Bad: "This was a stupid trade."
- Good: "This is a rule-adherence issue, not proof that the setup had no edge."
- Bad: "Never trade this pattern again."
## Incomplete Data
If a risk plan is missing:
- mark risk findings as `unclear`
- ask for the missing risk plan
- avoid claiming a rule violation
- still flag actual risk values if they are present
requirements.txt
# Runtime dependencies for standalone .skill installs (issue #330).
# Verified by scripts/check_skill_deps.py.
PyYAML>=6.0
scripts/review_trade_performance.py
#!/usr/bin/env python3
"""Deterministic trade performance coach.
Reviews closed trades / partial closes / monthly aggregates for process,
risk, execution, and possible behavior-pattern findings. This script is
advisory process-review tooling only; it does not produce trade signals or
broker instructions.
"""
from __future__ import annotations
import argparse
import json
import math
import re
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any
DISCLAIMER = (
"This review is for trading-process improvement only. It is not financial "
"advice, investment advice, therapy, mental-health diagnosis, or a trading "
"signal. The human trader remains responsible for all decisions and risk."
)
ALLOWED_ACTIONS = ["accept_rules", "modify_rules", "defer", "journal_only"]
@dataclass(frozen=True)
class Finding:
rule: str
status: str
evidence: str
severity: str
def to_dict(self) -> dict[str, str]:
return {
"rule": self.rule,
"status": self.status,
"evidence": self.evidence,
"severity": self.severity,
}
@dataclass(frozen=True)
class RiskNote:
topic: str
finding: str
severity: str
evidence: str
def to_dict(self) -> dict[str, str]:
return {
"topic": self.topic,
"finding": self.finding,
"severity": self.severity,
"evidence": self.evidence,
}
@dataclass(frozen=True)
class ExecutionFinding:
phase: str
finding: str
evidence: str
severity: str
def to_dict(self) -> dict[str, str]:
return {
"phase": self.phase,
"finding": self.finding,
"evidence": self.evidence,
"severity": self.severity,
}
@dataclass(frozen=True)
class BehaviorTag:
tag: str
confidence: str
evidence: str
reflection_question: str
def to_dict(self) -> dict[str, str]:
return {
"tag": self.tag,
"confidence": self.confidence,
"evidence": self.evidence,
"reflection_question": self.reflection_question,
}
def load_record(path: Path) -> dict[str, Any]:
"""Load JSON, or YAML if PyYAML is available."""
text = path.read_text(encoding="utf-8")
if path.suffix.lower() in {".yaml", ".yml"}:
try:
import yaml # type: ignore
except ImportError as exc: # pragma: no cover - environment dependent
raise RuntimeError(
"PyYAML is required for YAML input; use JSON or install pyyaml"
) from exc
data = yaml.safe_load(text)
else:
data = json.loads(text)
if not isinstance(data, dict):
raise ValueError(f"{path} must contain a JSON/YAML object")
return data
def deep_get(data: dict[str, Any], *keys: str, default: Any = None) -> Any:
current: Any = data
for key in keys:
if not isinstance(current, dict) or key not in current:
return default
current = current[key]
return current
def number(value: Any) -> float | None:
if value is None:
return None
if isinstance(value, bool):
return None
try:
exact = Decimal(str(value))
if not exact.is_finite() or exact < 0:
return None
result = float(value)
except (TypeError, ValueError, OverflowError, InvalidOperation):
return None
return result if math.isfinite(result) and result >= 0 else None
NUMERIC_FIELDS = (
("planned", "risk_r"),
("actual", "risk_r"),
("risk_plan", "max_risk_per_trade_r"),
("actual", "portfolio_heat_r"),
("risk_plan", "max_portfolio_heat_r"),
("monthly", "consecutive_losses"),
)
def validate_numeric_inputs(record: dict[str, Any]) -> None:
"""Reject invalid supplied evidence before it can look like missing data."""
for section, key in NUMERIC_FIELDS:
value = deep_get(record, section, key)
if value is None:
continue
parsed = number(value)
if parsed is None or (
key == "consecutive_losses"
and Decimal(str(value)) != Decimal(str(value)).to_integral_value()
):
requirement = "a finite nonnegative number"
if key == "consecutive_losses":
requirement = "a finite nonnegative whole number"
raise ValueError(f"{section}.{key} must be {requirement}; booleans are invalid")
def boolish(value: Any) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.strip().lower() in {"1", "true", "yes", "y"}
return bool(value)
def collect_text(record: dict[str, Any]) -> str:
parts: list[str] = []
for path in [
("journal", "reflection"),
("journal", "notes"),
("postmortem", "notes"),
("postmortem", "root_cause_notes"),
("planned", "thesis"),
("summary",),
]:
value = deep_get(record, *path)
if isinstance(value, list):
parts.extend(str(x) for x in value)
elif value:
parts.append(str(value))
emotions = deep_get(record, "journal", "emotions")
if isinstance(emotions, list):
parts.extend(str(x) for x in emotions)
return "\n".join(parts).lower()
def evaluate_process_adherence(record: dict[str, Any]) -> list[Finding]:
findings: list[Finding] = []
thesis_recorded = deep_get(record, "planned", "thesis_recorded_before_entry")
if thesis_recorded is False:
findings.append(
Finding(
"pre_entry_thesis_record",
"missed",
"planned.thesis_recorded_before_entry is false",
"warning",
)
)
elif thesis_recorded is None:
findings.append(
Finding(
"pre_entry_thesis_record",
"unclear",
"No explicit thesis_recorded_before_entry field was provided",
"warning",
)
)
else:
findings.append(
Finding(
"pre_entry_thesis_record",
"met",
"A pre-entry thesis record is present",
"info",
)
)
setup_confirmed = deep_get(record, "planned", "setup_confirmed")
entry_before_confirmation = boolish(
deep_get(record, "actual", "entry_before_confirmation", default=False)
)
if entry_before_confirmation or setup_confirmed is False:
findings.append(
Finding(
"setup_confirmation",
"missed",
"Entry appears to have occurred before planned setup confirmation",
"critical",
)
)
regime = str(deep_get(record, "planned", "market_regime", default="unknown")).lower()
traded_against_regime = boolish(
deep_get(record, "actual", "traded_against_regime", default=False)
)
if traded_against_regime or regime in {"restrictive", "cash_priority", "cash-priority"}:
findings.append(
Finding(
"market_regime_gate",
"missed",
f"Trade was taken while market_regime={regime!r} or actual.traded_against_regime=true",
"critical",
)
)
stop_moved = boolish(deep_get(record, "actual", "stop_moved", default=False))
stop_move_planned = boolish(deep_get(record, "actual", "stop_move_planned", default=False))
if stop_moved and not stop_move_planned:
findings.append(
Finding(
"stop_change_rule",
"missed",
"Stop was moved after entry without evidence of a pre-defined stop-change rule",
"critical",
)
)
return findings
def evaluate_risk_discipline(record: dict[str, Any]) -> list[RiskNote]:
notes: list[RiskNote] = []
actual_r = number(deep_get(record, "actual", "risk_r"))
planned_r = number(deep_get(record, "planned", "risk_r"))
max_r = number(deep_get(record, "risk_plan", "max_risk_per_trade_r"))
reference_r = max_r if max_r is not None else planned_r
if actual_r is not None and reference_r is not None:
if actual_r > reference_r * 1.25:
notes.append(
RiskNote(
"position_size",
"Actual risk materially exceeded the planned or maximum risk.",
"critical",
f"actual.risk_r={actual_r:g}, reference_r={reference_r:g}",
)
)
elif actual_r > reference_r:
notes.append(
RiskNote(
"position_size",
"Actual risk exceeded the plan by a small amount.",
"warning",
f"actual.risk_r={actual_r:g}, reference_r={reference_r:g}",
)
)
else:
notes.append(
RiskNote(
"position_size",
"Actual risk was within the provided risk limit.",
"info",
f"actual.risk_r={actual_r:g}, reference_r={reference_r:g}",
)
)
else:
notes.append(
RiskNote(
"position_size",
"Risk comparison is unclear because actual risk or risk plan is missing.",
"warning",
"Missing actual.risk_r, planned.risk_r, or risk_plan.max_risk_per_trade_r",
)
)
heat = number(deep_get(record, "actual", "portfolio_heat_r"))
max_heat = number(deep_get(record, "risk_plan", "max_portfolio_heat_r"))
if heat is not None and max_heat is not None:
if heat > max_heat:
notes.append(
RiskNote(
"portfolio_heat",
"Portfolio heat exceeded the provided maximum.",
"critical",
f"portfolio_heat_r={heat:g}, max_portfolio_heat_r={max_heat:g}",
)
)
elif heat > max_heat * 0.9:
notes.append(
RiskNote(
"portfolio_heat",
"Portfolio heat is close to the provided maximum.",
"warning",
f"portfolio_heat_r={heat:g}, max_portfolio_heat_r={max_heat:g}",
)
)
consecutive_losses = number(deep_get(record, "monthly", "consecutive_losses"))
if consecutive_losses is not None and consecutive_losses >= 3:
notes.append(
RiskNote(
"drawdown",
"Consecutive losses are elevated; consider temporary review-only mode.",
"critical" if consecutive_losses >= 4 else "warning",
f"monthly.consecutive_losses={consecutive_losses:g}",
)
)
return notes
def evaluate_execution_quality(record: dict[str, Any]) -> list[ExecutionFinding]:
findings: list[ExecutionFinding] = []
if boolish(deep_get(record, "actual", "entry_before_confirmation", default=False)):
findings.append(
ExecutionFinding(
"entry",
"Entry appears to have occurred before planned confirmation.",
"actual.entry_before_confirmation is true",
"critical",
)
)
else:
findings.append(
ExecutionFinding(
"entry",
"No explicit early-entry issue was detected from the provided record.",
"actual.entry_before_confirmation is false or absent",
"info",
)
)
if boolish(deep_get(record, "actual", "stop_moved", default=False)):
severity = (
"warning"
if boolish(deep_get(record, "actual", "stop_move_planned", default=False))
else "critical"
)
findings.append(
ExecutionFinding(
"stop",
"Stop was moved after entry."
if severity == "warning"
else "Stop was moved without a documented pre-defined rule.",
"actual.stop_moved is true",
severity,
)
)
if boolish(deep_get(record, "actual", "premature_exit", default=False)):
findings.append(
ExecutionFinding(
"exit",
"Exit appears to have occurred before planned invalidation or target rules.",
"actual.premature_exit is true",
"warning",
)
)
return findings
def has_any(text: str, patterns: list[str]) -> bool:
return any(re.search(pattern, text) for pattern in patterns)
def detect_behavioral_patterns(
record: dict[str, Any], findings: list[Finding], notes: list[RiskNote]
) -> list[BehaviorTag]:
text = collect_text(record)
tags: list[BehaviorTag] = []
def add(tag: str, confidence: str, evidence: str, question: str) -> None:
if tag not in {t.tag for t in tags}:
tags.append(BehaviorTag(tag, confidence, evidence, question))
if boolish(deep_get(record, "actual", "entry_before_confirmation", default=False)) or has_any(
text,
[r"didn'?t want to miss", r"fomo", r"chase", r"miss the move", r"moving fast"],
):
add(
"fomo_entry",
"medium" if "fomo" not in text else "high",
"Entry/journal evidence suggests fear of missing a move or entry before confirmation.",
"What evidence did you have before entry that would not have been available after waiting for confirmation?",
)
if has_any(text, [r"revenge", r"make it back", r"win it back", r"after.*loss"]):
add(
"revenge_trade",
"medium",
"Journal language suggests a possible attempt to recover prior losses.",
"Would you have taken this trade if the prior trade had been a winner?",
)
if boolish(deep_get(record, "actual", "premature_exit", default=False)) or has_any(
text, [r"got scared", r"took it off early", r"couldn'?t hold"]
):
add(
"premature_exit",
"medium",
"Exit/journal evidence suggests the trade may have been closed before planned invalidation or target rules.",
"What rule did the exit satisfy, or was it driven by discomfort?",
)
if has_any(text, [r"easy money", r"can'?t lose", r"sure thing", r"increased size after win"]):
add(
"overconfidence_after_winner",
"medium",
"Journal language suggests possible confidence escalation after a winner.",
"Did a prior win change the size or quality threshold for this trade?",
)
if boolish(deep_get(record, "actual", "stop_moved", default=False)) and not boolish(
deep_get(record, "actual", "stop_move_planned", default=False)
):
add(
"stop_moved",
"high",
"actual.stop_moved is true and no pre-planned stop-change rule was provided.",
"Was the new stop part of the plan before entry?",
)
# size_creep must require actual evidence that actual.risk_r > reference_r,
# not just any position_size warning (e.g. missing risk data also produces a warning,
# but that is not "size creep" — see Blocker 2 from the 2026-05-24 PR-F review).
actual_r = number(deep_get(record, "actual", "risk_r"))
max_r = number(deep_get(record, "risk_plan", "max_risk_per_trade_r"))
planned_r = number(deep_get(record, "planned", "risk_r"))
reference_r = max_r if max_r is not None else planned_r
if actual_r is not None and reference_r is not None and actual_r > reference_r:
add(
"size_creep",
"high",
f"Actual risk exceeded the planned or maximum risk (actual.risk_r={actual_r:g} > reference_r={reference_r:g}).",
"What would this trade have looked like at the planned risk size?",
)
elif any(
n.topic == "position_size" and n.severity in {"warning", "critical"} for n in notes
) and (actual_r is None or reference_r is None):
add(
"unknown_size_discipline",
"medium",
"Risk discipline could not be assessed because actual.risk_r or risk_plan/planned risk reference is missing.",
"Can the risk plan or actual risk be recorded for the next trade so size discipline is verifiable?",
)
if boolish(deep_get(record, "actual", "hesitated", default=False)) or has_any(
text, [r"hesitat", r"froze", r"missed my entry"]
):
add(
"hesitation",
"medium",
"Journal/action evidence suggests planned execution may have been delayed or missed.",
"What objective trigger would have removed the need for discretion?",
)
if len([f for f in findings if f.severity in {"warning", "critical"}]) >= 3:
add(
"rule_drift",
"medium",
"Multiple process findings suggest possible rule drift.",
"Which single rule should be simplified and enforced next session?",
)
if not tags:
tags.append(
BehaviorTag(
"no_pattern_detected",
"low",
"No behavior pattern was detected from the available evidence.",
"Is there any additional journal context that should be added before finalizing this review?",
)
)
return tags
def compute_score(findings: list[Any]) -> int:
score = 100
for item in findings:
severity = getattr(item, "severity", "info")
if severity == "critical":
score -= 30
elif severity == "warning":
score -= 15
elif severity == "info":
score -= 0
return max(0, min(100, score))
def determine_verdict(
findings: list[Finding],
notes: list[RiskNote],
tags: list[BehaviorTag],
record: dict[str, Any],
execution: list[ExecutionFinding] | None = None,
) -> str:
# Blocker 1 (2026-05-24 PR-F review): execution warnings/critical must also escalate
# the verdict — previously a premature_exit produced an execution finding + behavior
# tag but the verdict still came out OK because only process/risk were inspected.
all_severities = [*findings, *notes, *(execution or [])]
critical_count = sum(1 for x in all_severities if getattr(x, "severity", "") == "critical")
revenge = any(t.tag == "revenge_trade" for t in tags)
consecutive_losses = number(deep_get(record, "monthly", "consecutive_losses")) or 0
if critical_count >= 2 or (revenge and consecutive_losses >= 2) or consecutive_losses >= 4:
return "COOL_DOWN"
if critical_count >= 1:
return "RULE_VIOLATION"
if any(getattr(x, "severity", "") == "warning" for x in all_severities):
return "REVIEW_REQUIRED"
return "OK"
def infer_root_cause(record: dict[str, Any], findings: list[Finding], notes: list[RiskNote]) -> str:
provided = deep_get(record, "postmortem", "root_cause")
if provided:
return str(provided)
if any(n.topic == "position_size" and n.severity in {"warning", "critical"} for n in notes):
return "risk_sizing"
if any(
f.rule in {"setup_confirmation", "stop_change_rule"} and f.severity == "critical"
for f in findings
):
return "execution"
if any(f.rule == "market_regime_gate" and f.severity == "critical" for f in findings):
return "market_environment"
return "unknown"
def next_session_rules(
verdict: str, notes: list[RiskNote], tags: list[BehaviorTag]
) -> list[dict[str, str]]:
rules: list[dict[str, str]] = []
def add(rule: str, duration: str, trigger: str, reason: str) -> None:
if rule not in {r["rule"] for r in rules}:
rules.append({"rule": rule, "duration": duration, "trigger": trigger, "reason": reason})
# Blocker 2 follow-up (2026-05-24 PR-F review): the cap-risk rule must
# only fire when there is explicit evidence that actual risk exceeded the
# plan (size_creep tag). The missing-risk-data case (unknown_size_discipline)
# is "unverifiable", not "exceeded", so it gets a different rule asking the
# trader to record planned/actual risk next time.
if any(t.tag == "size_creep" for t in tags):
add(
"Cap risk at 0.5R for the next two trades unless the trader explicitly modifies the risk plan in the journal.",
"next_two_trades",
"size_creep tag",
"Actual risk exceeded the stated risk plan.",
)
elif any(t.tag == "unknown_size_discipline" for t in tags):
add(
"On the next trade, record both planned risk_r and actual risk_r in the journal so risk discipline becomes verifiable.",
"next_trade",
"unknown_size_discipline tag",
"Risk discipline could not be assessed because planned or actual risk was missing.",
)
if any(t.tag == "fomo_entry" for t in tags):
add(
"No new entry without a pre-entry thesis record, invalidation point, and setup-confirmation note.",
"next_week",
"possible FOMO entry pattern",
"The review found evidence of entry before confirmation or fear of missing a move.",
)
if any(t.tag == "revenge_trade" for t in tags) or verdict == "COOL_DOWN":
add(
"Switch to review-only mode for the next session; do not add new risk until the rule review is journaled.",
"next_session",
"possible revenge/cool-down condition",
"Repeated losses or revenge-trade evidence can escalate risk-taking.",
)
if any(t.tag == "stop_moved" for t in tags):
add(
"Any stop adjustment must be written before entry; unplanned stop changes require immediate post-trade review.",
"next_month",
"unplanned stop move",
"Stop discipline protects the loss limit and keeps outcomes reviewable.",
)
if not rules:
add(
"Keep the current process rules unchanged and continue journaling every trade outcome.",
"next_session",
"no material process issue detected",
"The available evidence did not show a rule violation.",
)
return rules
def build_review(record: dict[str, Any], source_records: list[str]) -> dict[str, Any]:
validate_numeric_inputs(record)
process = evaluate_process_adherence(record)
risk_notes = evaluate_risk_discipline(record)
execution = evaluate_execution_quality(record)
tags = detect_behavioral_patterns(record, process, risk_notes)
verdict = determine_verdict(process, risk_notes, tags, record, execution=execution)
root_cause = infer_root_cause(record, process, risk_notes)
review_id = str(
record.get("review_id")
or record.get("trade_id")
or f"review_{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
)
review_type = str(record.get("review_type") or "single_trade")
outcome = str(
record.get("outcome") or deep_get(record, "summary", "outcome", default="unknown")
)
scores = {
"process_score": compute_score(process),
"risk_score": compute_score(risk_notes),
"execution_score": compute_score(execution),
"review_quality_score": review_quality_score(record),
}
return {
"schema_version": 1,
"review_type": review_type,
"review_id": review_id,
"generated_at": datetime.now(timezone.utc).isoformat(),
"source_records": source_records,
"overall_verdict": verdict,
"summary": {
"outcome": outcome,
"primary_root_cause": root_cause,
"secondary_root_causes": infer_secondary_causes(process, risk_notes, tags),
"confidence": infer_confidence(scores["review_quality_score"], process, risk_notes),
},
"scores": scores,
"process_adherence_findings": [f.to_dict() for f in process],
"risk_manager_notes": [n.to_dict() for n in risk_notes],
"execution_quality_assessment": [e.to_dict() for e in execution],
"behavioral_pattern_tags": [t.to_dict() for t in tags],
"next_session_operating_rules": next_session_rules(verdict, risk_notes, tags),
"coach_questions": [t.reflection_question for t in tags[:4]],
"human_decision_gate": {
"question": "Do you accept these temporary operating rules, modify them, defer the decision, or journal this review only?",
"allowed_actions": ALLOWED_ACTIONS,
"default_action": "journal_only",
},
"disclaimer": DISCLAIMER,
}
def review_quality_score(record: dict[str, Any]) -> int:
expected = [
("planned", "risk_r"),
("actual", "risk_r"),
("risk_plan", "max_risk_per_trade_r"),
("postmortem", "root_cause"),
("journal", "reflection"),
]
present = sum(1 for path in expected if deep_get(record, *path) is not None)
return int(40 + 60 * (present / len(expected)))
def infer_secondary_causes(
process: list[Finding], notes: list[RiskNote], tags: list[BehaviorTag]
) -> list[str]:
causes: list[str] = []
if any(f.rule == "market_regime_gate" and f.severity == "critical" for f in process):
causes.append("market_environment")
if any(n.topic == "position_size" and n.severity in {"warning", "critical"} for n in notes):
causes.append("risk_sizing")
if any(t.tag in {"fomo_entry", "revenge_trade", "stop_moved"} for t in tags):
causes.append("behavior_pattern")
return list(dict.fromkeys(causes))
def infer_confidence(review_quality: int, process: list[Finding], notes: list[RiskNote]) -> str:
if review_quality < 60:
return "low"
if review_quality >= 85 and (process or notes):
return "high"
return "medium"
def render_markdown(report: dict[str, Any]) -> str:
lines = [
f"# Trade Performance Coach Report: {report['review_id']}",
"",
f"**Verdict:** `{report['overall_verdict']}`",
f"**Review type:** `{report['review_type']}`",
f"**Primary root cause:** `{report['summary']['primary_root_cause']}`",
"",
"## Scores",
"",
]
for key, value in report["scores"].items():
lines.append(f"- `{key}`: {value}")
lines.extend(["", "## Process Adherence", ""])
for item in report["process_adherence_findings"]:
lines.append(
f"- **{item['severity'].upper()}** `{item['rule']}` — {item['status']}: {item['evidence']}"
)
lines.extend(["", "## Risk Manager Notes", ""])
for item in report["risk_manager_notes"]:
lines.append(
f"- **{item['severity'].upper()}** `{item['topic']}` — {item['finding']} Evidence: {item['evidence']}"
)
lines.extend(["", "## Execution Quality", ""])
for item in report["execution_quality_assessment"]:
lines.append(
f"- **{item['severity'].upper()}** `{item['phase']}` — {item['finding']} Evidence: {item['evidence']}"
)
lines.extend(["", "## Possible Behavioral Patterns", ""])
for item in report["behavioral_pattern_tags"]:
lines.append(f"- `{item['tag']}` ({item['confidence']}): {item['evidence']}")
lines.append(f" - Reflection: {item['reflection_question']}")
lines.extend(["", "## Next-Session Operating Rules", ""])
for item in report["next_session_operating_rules"]:
lines.append(f"- **{item['duration']}**: {item['rule']}")
lines.append(f" - Trigger: {item['trigger']}")
lines.append(f" - Reason: {item['reason']}")
lines.extend(["", "## Coach Questions", ""])
for question in report["coach_questions"]:
lines.append(f"- {question}")
gate = report["human_decision_gate"]
lines.extend(
[
"",
"## Human Decision Gate",
"",
gate["question"],
"",
f"Allowed actions: {', '.join(gate['allowed_actions'])}",
f"Default action: `{gate['default_action']}`",
"",
"## Disclaimer",
"",
report["disclaimer"],
"",
]
)
return "\n".join(lines)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Review trade performance records.")
parser.add_argument(
"--input",
action="append",
required=True,
help="Input JSON/YAML record. Can be passed multiple times.",
)
parser.add_argument(
"--output-dir",
default="reports/trade-performance-coach",
help="Directory for output reports.",
)
parser.add_argument("--json-name", default=None, help="Optional JSON output filename.")
parser.add_argument("--markdown", action="store_true", help="Also write Markdown report.")
parser.add_argument("--stdout", action="store_true", help="Print JSON report to stdout.")
args = parser.parse_args(argv)
input_paths = [Path(p) for p in args.input]
try:
records = [load_record(path) for path in input_paths]
for source_record in records:
validate_numeric_inputs(source_record)
except ValueError as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
if len(input_paths) == 1:
record = records[0]
else:
# Medium 1 (2026-05-24 PR-F review): multi-input mode currently wraps records
# without aggregating process/risk/text per trade — the analysis is therefore
# shallow. Warn loudly so users supply a pre-aggregated monthly JSON when they
# want accurate monthly analysis. A full aggregate_monthly_inputs() implementation
# is tracked as follow-up.
print(
"warning: multi-input mode wraps records without aggregating process/risk/text per trade. "
"Pass a single pre-aggregated monthly JSON (with monthly.trades_summary, etc.) for accurate "
"monthly analysis. See follow-up: trade-performance-coach: implement multi-input aggregate logic.",
file=sys.stderr,
)
# Monthly aggregate wrapper for multiple records.
record = {
"review_type": "monthly_aggregate",
"trade_id": "monthly_aggregate",
"outcome": "mixed",
"monthly": {"trades": records},
"journal": {"reflection": "Multiple records supplied for aggregate review."},
}
report = build_review(record, [p.as_posix() for p in input_paths])
out_dir = Path(args.output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
json_name = args.json_name or f"trade_performance_coach_{report['review_id']}.json"
json_path = out_dir / json_name
json_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
if args.markdown:
md_path = out_dir / json_name.replace(".json", ".md")
md_path.write_text(render_markdown(report), encoding="utf-8")
if args.stdout:
print(json.dumps(report, indent=2, sort_keys=True))
else:
print(json_path)
return 0
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())
scripts/tests/fixtures/incomplete_record.json
{
"schema_version": 1,
"review_type": "single_trade",
"trade_id": "incomplete_EXMPL_001",
"ticker": "EXMPL",
"outcome": "unknown",
"actual": {
"entry": 100.0
},
"journal": {
"reflection": "I need to add more details later."
}
}
scripts/tests/fixtures/monthly_aggregate_revenge_pattern.json
{
"schema_version": 1,
"review_type": "monthly_aggregate",
"trade_id": "monthly_2026_05",
"outcome": "mixed",
"planned": {
"thesis_recorded_before_entry": true,
"setup_confirmed": true,
"market_regime": "allowed"
},
"actual": {
"risk_r": 1.4,
"portfolio_heat_r": 3.7,
"stop_moved": false,
"entry_before_confirmation": false,
"traded_against_regime": false
},
"risk_plan": {
"max_risk_per_trade_r": 1.0,
"max_portfolio_heat_r": 4.0
},
"monthly": {
"trades": ["EXMPL1", "EXMPL2", "EXMPL3"],
"consecutive_losses": 3,
"rule_violations": 2
},
"postmortem": {
"root_cause": "risk_sizing",
"notes": ["After two losses, I increased size to make it back."]
},
"journal": {
"reflection": "I wanted to make it back quickly after the prior loss.",
"emotions": ["frustrated", "revenge"]
}
}
scripts/tests/fixtures/partial_close_stop_moved.json
{
"schema_version": 1,
"review_type": "partial_close",
"trade_id": "partial_EXMPL_001",
"ticker": "EXMPL",
"outcome": "open",
"planned": {
"thesis": "Trend continuation with partial at 2R.",
"entry": 75.0,
"stop": 70.0,
"target": 85.0,
"risk_r": 1.0,
"thesis_recorded_before_entry": true,
"setup_confirmed": true,
"market_regime": "allowed"
},
"actual": {
"entry": 75.2,
"partial_exit": 79.5,
"risk_r": 1.0,
"portfolio_heat_r": 3.2,
"stop_moved": true,
"stop_move_planned": false,
"entry_before_confirmation": false,
"traded_against_regime": false
},
"risk_plan": {
"max_risk_per_trade_r": 1.0,
"max_portfolio_heat_r": 4.0
},
"postmortem": {
"root_cause": "execution",
"notes": ["Partial was fine, but I moved the stop manually without a written rule."]
},
"journal": {
"reflection": "I moved the stop because I was worried about giving back gains.",
"emotions": ["anxious"]
}
}
scripts/tests/fixtures/risk_data_missing_no_size_creep.json
{
"schema_version": 1,
"review_type": "single_trade",
"trade_id": "missing_risk_EXMPL_001",
"ticker": "EXMPL",
"outcome": "loss",
"planned": {
"thesis": "Stage 2 setup; risk plan was not recorded in the broker-side ticket.",
"entry": 50.0,
"stop": 47.0,
"target": 56.0,
"thesis_recorded_before_entry": true,
"setup_confirmed": true,
"market_regime": "allowed"
},
"actual": {
"entry": 50.0,
"exit": 47.0,
"stop_moved": false,
"entry_before_confirmation": false,
"traded_against_regime": false
},
"postmortem": {
"notes": [
"Risk plan and actual risk sizing fields are missing from this record (legacy broker import). The trade closed at the planned stop. There is no evidence of size creep."
]
},
"journal": {
"reflection": "Followed the plan; the stop hit.",
"emotions": ["calm"]
}
}
scripts/tests/fixtures/single_trade_clean_loss.json
{
"schema_version": 1,
"review_type": "single_trade",
"trade_id": "clean_loss_EXMPL_001",
"ticker": "EXMPL",
"outcome": "loss",
"planned": {
"thesis": "Stage 2 pullback continuation setup with defined invalidation.",
"entry": 100.0,
"stop": 95.0,
"target": 112.0,
"risk_r": 1.0,
"thesis_recorded_before_entry": true,
"setup_confirmed": true,
"market_regime": "allowed"
},
"actual": {
"entry": 100.5,
"exit": 95.0,
"risk_r": 1.0,
"portfolio_heat_r": 2.5,
"stop_moved": false,
"entry_before_confirmation": false,
"traded_against_regime": false
},
"risk_plan": {
"max_risk_per_trade_r": 1.0,
"max_portfolio_heat_r": 4.0
},
"postmortem": {
"root_cause": "randomness",
"notes": ["Setup was valid. The market reversed after entry and stop was hit as planned."]
},
"journal": {
"reflection": "I followed the plan and accepted the loss.",
"emotions": ["calm"]
}
}
scripts/tests/fixtures/single_trade_premature_exit.json
{
"schema_version": 1,
"review_type": "single_trade",
"trade_id": "premature_exit_EXMPL_001",
"ticker": "EXMPL",
"outcome": "loss",
"planned": {
"thesis": "Stage 2 breakout setup with multi-week base; target = +2R measured move.",
"entry": 100.0,
"stop": 97.0,
"target": 106.0,
"risk_r": 1.0,
"thesis_recorded_before_entry": true,
"setup_confirmed": true,
"market_regime": "allowed"
},
"actual": {
"entry": 100.2,
"exit": 99.5,
"risk_r": 1.0,
"portfolio_heat_r": 1.5,
"stop_moved": false,
"entry_before_confirmation": false,
"traded_against_regime": false,
"premature_exit": true
},
"risk_plan": {
"max_risk_per_trade_r": 1.0,
"max_portfolio_heat_r": 4.0
},
"postmortem": {
"notes": [
"Trade was working but I closed it before the stop and before the target was approached."
]
},
"journal": {
"reflection": "Got scared and took it off early. Stop was never touched.",
"emotions": ["anxious"]
}
}
scripts/tests/fixtures/single_trade_rule_violation_loss.json
{
"schema_version": 1,
"review_type": "single_trade",
"trade_id": "rule_violation_EXMPL_001",
"ticker": "EXMPL",
"outcome": "loss",
"planned": {
"thesis": "Breakout continuation setup, but confirmation was not complete.",
"entry": 50.0,
"stop": 47.5,
"target": 57.5,
"risk_r": 1.0,
"thesis_recorded_before_entry": false,
"setup_confirmed": false,
"market_regime": "restrictive"
},
"actual": {
"entry": 54.0,
"exit": 46.0,
"risk_r": 1.8,
"portfolio_heat_r": 4.2,
"stop_moved": true,
"stop_move_planned": false,
"entry_before_confirmation": true,
"traded_against_regime": true
},
"risk_plan": {
"max_risk_per_trade_r": 1.0,
"max_portfolio_heat_r": 4.0
},
"postmortem": {
"root_cause": "rule_violation",
"notes": ["I entered before confirmation and moved the stop after the trade went against me."]
},
"journal": {
"reflection": "I did not want to miss the move because the stock was moving fast.",
"emotions": ["FOMO", "frustrated"]
}
}
scripts/tests/test_review_trade_performance.py
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
SCRIPT_DIR = Path(__file__).resolve().parents[1]
FIXTURES = Path(__file__).resolve().parent / "fixtures"
sys.path.insert(0, str(SCRIPT_DIR))
import review_trade_performance as rpc # noqa: E402
def load(name: str) -> dict:
return json.loads((FIXTURES / name).read_text(encoding="utf-8"))
def test_clean_process_loss_is_ok_or_warn_without_behavior_tag():
report = rpc.build_review(load("single_trade_clean_loss.json"), ["fixture"])
assert report["overall_verdict"] == "OK"
assert report["summary"]["primary_root_cause"] == "randomness"
assert report["scores"]["risk_score"] == 100
tags = {t["tag"] for t in report["behavioral_pattern_tags"]}
assert tags == {"no_pattern_detected"}
assert "financial advice" in report["disclaimer"].lower()
def test_rule_violation_loss_detects_fomo_size_creep_and_stop_moved():
report = rpc.build_review(load("single_trade_rule_violation_loss.json"), ["fixture"])
assert report["overall_verdict"] == "COOL_DOWN"
tags = {t["tag"] for t in report["behavioral_pattern_tags"]}
assert {"fomo_entry", "size_creep", "stop_moved"}.issubset(tags)
process_rules = {f["rule"]: f for f in report["process_adherence_findings"]}
assert process_rules["setup_confirmation"]["severity"] == "critical"
assert process_rules["market_regime_gate"]["severity"] == "critical"
assert any(
n["topic"] == "position_size" and n["severity"] == "critical"
for n in report["risk_manager_notes"]
)
assert any("0.5R" in r["rule"] for r in report["next_session_operating_rules"])
def test_partial_close_stop_moved_detects_stop_pattern():
report = rpc.build_review(load("partial_close_stop_moved.json"), ["fixture"])
assert report["review_type"] == "partial_close"
# Note (2026-05-24 PR-F Blocker 1 fix): execution_quality_assessment now contributes
# to determine_verdict. An unplanned stop move surfaces as critical in both the
# process_adherence (stop_change_rule) and execution_quality (stop phase) axes,
# so critical_count reaches 2 and the verdict escalates to COOL_DOWN. This is
# conservative-side correct — stop_moved without a pre-defined rule is materially
# serious — so COOL_DOWN is included alongside RULE_VIOLATION / REVIEW_REQUIRED here.
assert report["overall_verdict"] in {"RULE_VIOLATION", "REVIEW_REQUIRED", "COOL_DOWN"}
tags = {t["tag"] for t in report["behavioral_pattern_tags"]}
assert "stop_moved" in tags
assert any(e["phase"] == "stop" for e in report["execution_quality_assessment"])
def test_monthly_aggregate_revenge_pattern_cool_down():
report = rpc.build_review(load("monthly_aggregate_revenge_pattern.json"), ["fixture"])
assert report["review_type"] == "monthly_aggregate"
assert report["overall_verdict"] == "COOL_DOWN"
tags = {t["tag"] for t in report["behavioral_pattern_tags"]}
assert "revenge_trade" in tags
assert any("review-only" in r["rule"].lower() for r in report["next_session_operating_rules"])
def test_incomplete_record_degrades_without_crashing():
report = rpc.build_review(load("incomplete_record.json"), ["fixture"])
assert report["overall_verdict"] in {"REVIEW_REQUIRED", "WARN"}
assert report["scores"]["review_quality_score"] < 80
assert any(f["status"] == "unclear" for f in report["process_adherence_findings"])
assert report["human_decision_gate"]["default_action"] == "journal_only"
def test_premature_exit_is_not_verdict_ok():
"""Blocker 1 regression (2026-05-24 PR-F review):
actual.premature_exit=true must escalate the verdict beyond OK
(execution warnings now contribute to determine_verdict).
"""
report = rpc.build_review(load("single_trade_premature_exit.json"), ["fixture"])
assert report["overall_verdict"] != "OK"
assert report["overall_verdict"] in {"REVIEW_REQUIRED", "WARN", "RULE_VIOLATION", "COOL_DOWN"}
assert any(
e["phase"] == "exit" and e["severity"] == "warning"
for e in report["execution_quality_assessment"]
)
tags = {t["tag"] for t in report["behavioral_pattern_tags"]}
assert "premature_exit" in tags
def test_missing_risk_data_does_not_tag_size_creep():
"""Blocker 2 regression (2026-05-24 PR-F review):
size_creep must require explicit evidence that actual.risk_r > reference_r.
A missing-risk-data position_size warning must NOT be tagged as size_creep;
instead it should surface as unknown_size_discipline. The next-session rule
must NOT claim "actual risk exceeded the plan" (which would contradict the
missing-data state); it must ask the trader to record planned/actual risk
next time.
"""
report = rpc.build_review(load("risk_data_missing_no_size_creep.json"), ["fixture"])
tags = {t["tag"] for t in report["behavioral_pattern_tags"]}
assert "size_creep" not in tags
assert "unknown_size_discipline" in tags
assert any(
n["topic"] == "position_size" and n["severity"] == "warning"
for n in report["risk_manager_notes"]
)
# The Cap-risk-at-0.5R rule (which states "Actual risk exceeded the stated
# risk plan") must NOT fire when risk discipline is unverifiable.
rules = report["next_session_operating_rules"]
assert all("0.5R" not in r["rule"] for r in rules), (
"Cap-risk-at-0.5R rule must only fire for size_creep, not unknown_size_discipline"
)
assert all("exceeded" not in r["reason"].lower() for r in rules), (
"next-session rule reason must not claim risk exceeded when data is missing"
)
# Instead, a rule asking the trader to record planned/actual risk must fire.
assert any("record" in r["rule"].lower() and "risk" in r["rule"].lower() for r in rules), (
"unknown_size_discipline must produce a record-planned/actual-risk rule"
)
def test_report_validates_against_schema():
"""Schema enum coverage regression (2026-05-24 PR-F review):
every behavior_tag the runtime can emit must be in the schema's enum,
including unknown_size_discipline. Validate one report per fixture so
that adding a new tag without updating the schema fails this test.
"""
try:
import jsonschema # type: ignore
except ImportError: # pragma: no cover - environment dependent
import pytest
pytest.skip("jsonschema not installed")
schema_path = SCRIPT_DIR.parent / "assets" / "performance_coach_report.schema.json"
schema = json.loads(schema_path.read_text(encoding="utf-8"))
fixtures = [
"single_trade_clean_loss.json",
"single_trade_rule_violation_loss.json",
"partial_close_stop_moved.json",
"monthly_aggregate_revenge_pattern.json",
"incomplete_record.json",
"single_trade_premature_exit.json",
"risk_data_missing_no_size_creep.json",
]
for fixture in fixtures:
report = rpc.build_review(load(fixture), [fixture])
# schema_version is int per dataclass output; the asset schema uses 1.0
# as illustrative — coerce to match the structural enum check below.
jsonschema.Draft7Validator(schema).validate(report)
def test_cli_writes_json_and_markdown(tmp_path):
fixture = FIXTURES / "single_trade_rule_violation_loss.json"
rc = rpc.main(
[
"--input",
str(fixture),
"--output-dir",
str(tmp_path),
"--json-name",
"report.json",
"--markdown",
]
)
assert rc == 0
report_json = tmp_path / "report.json"
report_md = tmp_path / "report.md"
assert report_json.exists()
assert report_md.exists()
data = json.loads(report_json.read_text(encoding="utf-8"))
assert data["overall_verdict"] == "COOL_DOWN"
assert "Human Decision Gate" in report_md.read_text(encoding="utf-8")
NUMERIC_FIELDS = [
("planned", "risk_r"),
("actual", "risk_r"),
("risk_plan", "max_risk_per_trade_r"),
("actual", "portfolio_heat_r"),
("risk_plan", "max_portfolio_heat_r"),
("monthly", "consecutive_losses"),
]
@pytest.mark.parametrize("section,key", NUMERIC_FIELDS)
@pytest.mark.parametrize(
"value",
[float("nan"), float("inf"), -float("inf"), -1, True, "bad", "1e999", "-1e-999", 10**400],
)
def test_invalid_numeric_evidence_rejected(section, key, value):
record = load("single_trade_clean_loss.json")
record.setdefault(section, {})[key] = value
with pytest.raises(ValueError, match=rf"{section}\.{key}"):
rpc.build_review(record, ["fixture"])
@pytest.mark.parametrize("value", [0.5, "2.5", "2.0000000000000001"])
def test_fractional_loss_count_rejected(value):
record = load("single_trade_clean_loss.json")
record["monthly"] = {"consecutive_losses": value}
with pytest.raises(ValueError, match="monthly.consecutive_losses"):
rpc.build_review(record, ["fixture"])
@pytest.mark.parametrize("value", [0, "0", "1.0"])
def test_valid_nonnegative_numeric_evidence(value):
record = load("single_trade_clean_loss.json")
for section, key in NUMERIC_FIELDS:
record.setdefault(section, {})[key] = value
report = rpc.build_review(record, ["fixture"])
assert report["overall_verdict"] in {"OK", "REVIEW_REQUIRED"}
assert report["risk_manager_notes"][0]["severity"] == "info"
def test_null_optional_maximum_keeps_planned_risk_fallback():
record = load("single_trade_clean_loss.json")
record["risk_plan"]["max_risk_per_trade_r"] = None
assert rpc.build_review(record, ["fixture"])["overall_verdict"] == "OK"
def test_extreme_finite_risk_comparison_remains_warning():
record = load("single_trade_clean_loss.json")
record["actual"]["risk_r"] = 1.7e308
record["risk_plan"]["max_risk_per_trade_r"] = 1.5e308
report = rpc.build_review(record, ["fixture"])
assert report["overall_verdict"] == "REVIEW_REQUIRED"
assert report["risk_manager_notes"][0]["severity"] == "warning"
@pytest.mark.parametrize("suffix", ["json", "yaml"])
@pytest.mark.parametrize("multiple", [False, True])
@pytest.mark.parametrize("existing", [False, True])
def test_invalid_cli_input_leaves_outputs_untouched(tmp_path, capsys, suffix, multiple, existing):
record = load("single_trade_clean_loss.json")
record["risk_plan"]["max_risk_per_trade_r"] = "NaN"
source = tmp_path / f"invalid.{suffix}"
if suffix == "yaml":
import yaml
source.write_text(yaml.safe_dump(record), encoding="utf-8")
else:
source.write_text(json.dumps(record), encoding="utf-8")
out = tmp_path / "output"
if existing:
out.mkdir()
(out / "report.json").write_text("existing report", encoding="utf-8")
args = [
"--input",
str(source),
"--output-dir",
str(out),
"--json-name",
"report.json",
"--markdown",
]
if multiple:
args[0:0] = ["--input", str(FIXTURES / "single_trade_clean_loss.json")]
assert rpc.main(args) == 2
assert "risk_plan.max_risk_per_trade_r" in capsys.readouterr().err
if existing:
assert list(out.iterdir()) == [out / "report.json"]
assert (out / "report.json").read_text() == "existing report"
else:
assert not out.exists()
def test_exact_125_percent_risk_is_warning_not_critical():
record = load("single_trade_clean_loss.json")
record["actual"]["risk_r"] = 0.75
record["risk_plan"]["max_risk_per_trade_r"] = 0.6
report = rpc.build_review(record, ["fixture"])
assert report["risk_manager_notes"][0]["severity"] == "warning"
SKILL.md
---
name: trade-performance-coach
description: >-
Review closed trades, partial exits, and monthly trade aggregates for process
adherence, risk discipline, execution quality, and evidence-based trading
behavior patterns. Use after trader-memory-core and signal-postmortem have
produced records, or when the user asks for a post-trade coach, risk-manager
style review, rule-adherence review, next-session operating rules, or
psychology-aware trading behavior feedback. This skill does not provide buy/sell
advice, therapy, or broker execution.
---
# Trade Performance Coach
## Overview
Trade Performance Coach reviews recorded trade outcomes and journal evidence to
help a human trader improve their decision process. It converts closed-trade
records, postmortem findings, risk rules, and optional market-regime context into
an evidence-based coaching report covering:
- process adherence
- risk discipline
- execution quality
- possible trading-behavior patterns
- next-session operating rules
- coach questions for reflection
This skill is intended to fill the support role that a risk manager, desk lead,
or trading coach might provide in a professional trading environment. It is
strictly a process-review skill: it never recommends entering, exiting, buying,
selling, shorting, holding, or sizing a specific security.
## When to Use
Use this skill when any of the following are true:
- A trade has been closed and the user wants a post-trade coaching review.
- A partial close occurred and the user wants to inspect sizing, stop, or exit behavior.
- The user has `trader-memory-core` thesis records and `signal-postmortem` findings and wants next-session operating rules.
- The user wants a monthly review of recurring process, risk, execution, or behavior patterns.
- The user asks for a risk-manager style review of their own recorded trades.
- The user asks whether a loss was a process error, execution error, market environment issue, or acceptable variance.
- The user wants possible FOMO, revenge-trade, overconfidence, hesitation, stop-moving, or size-creep patterns flagged with evidence.
## When Not to Use
Do not use this skill to:
- Pick stocks or rank trade candidates.
- Approve or reject a live trade as financial advice.
- Place orders or draft broker instructions.
- Provide therapy, mental-health diagnosis, or personality assessment.
- Infer private psychological traits beyond the trade evidence supplied.
- Shame the user for losses or rule violations.
- Replace `trader-memory-core`; this skill consumes journal/thesis records and produces coaching findings.
If the input is incomplete, default to `REVIEW_REQUIRED` or `journal_only` mode and ask for missing records rather than inventing evidence.
## Prerequisites
Recommended upstream records:
- `trader-memory-core` closed thesis record or journal entry
- `signal-postmortem` postmortem findings
- original trade plan or trade ticket
- actual entry / exit / partial-close actions
- user-defined risk plan, if available
- optional `market-regime-daily` / `exposure-coach` context
No paid API key is required. The deterministic script works from local JSON/YAML-like records.
## Inputs
Minimum useful input is one recorded trade or one monthly aggregate.
Preferred fields:
```yaml
review_type: single_trade | partial_close | monthly_aggregate
trade_id: string
ticker: string
outcome: win | loss | breakeven | mixed
planned:
thesis: string
entry: number
stop: number
target: number
risk_r: number
thesis_recorded_before_entry: boolean
setup_confirmed: boolean
market_regime: allowed | restrictive | cash_priority | unknown
actual:
entry: number
exit: number
risk_r: number
portfolio_heat_r: number
stop_moved: boolean
stop_move_planned: boolean
entry_before_confirmation: boolean
traded_against_regime: boolean
risk_plan:
max_risk_per_trade_r: number
max_portfolio_heat_r: number
max_weekly_loss_r: number
postmortem:
root_cause: thesis_quality | execution | risk_sizing | market_environment | rule_violation | randomness | unknown
notes: [string]
journal:
reflection: string
emotions: [string]
monthly:
trades: [object]
consecutive_losses: number
rule_violations: number
```
The script tolerates partial records. Missing evidence is marked as `unclear`.
For the numeric fields actually evaluated (`planned.risk_r`, `actual.risk_r`,
`risk_plan.max_risk_per_trade_r`, `actual.portfolio_heat_r`,
`risk_plan.max_portfolio_heat_r`, and `monthly.consecutive_losses`), supplied
non-null values must be finite and nonnegative. Numeric strings and zero are
accepted; consecutive losses must be a whole number. Booleans, negative values,
NaN, infinity, malformed strings, and conversion overflow are rejected. An
explicitly invalid maximum never falls back to planned risk. Missing/null fields
retain the partial-record behavior. The CLI validates every source record,
including multiple inputs, and returns exit code 2 with a field-specific error
before creating or modifying reports when a numeric value is invalid.
This skill remains beta. Numeric validation does not establish production
readiness; report-ID path safety and the documented shallow multi-input wrapper
still require separate assessment.
## Workflow
### Step 1 — Collect source records
Collect the most recent closed trade record, postmortem, risk plan, and journal notes.
```bash
python3 skills/trade-performance-coach/scripts/review_trade_performance.py \
--input reports/trade_memory/closed_thesis_EXMPL.json \
--output-dir reports/trade-performance-coach
```
### Step 2 — Evaluate process adherence
Compare actual actions against the user's documented plan and rules. Check for:
- missing pre-entry thesis
- setup confirmation skipped
- trade taken against market-regime gate
- stop moved without a pre-defined rule
- exit / partial close inconsistent with plan
- incomplete record quality
### Step 3 — Evaluate risk discipline
Compare actual risk and heat against the risk plan. Check for:
- per-trade risk above max
- portfolio heat above max
- weekly loss or consecutive-loss escalation
- oversized trade after a winner or loser
- correlated exposure if provided
### Step 4 — Evaluate execution quality
Classify entry, stop, exit, add, trim, and review behavior. Separate clean-process losses from execution mistakes.
### Step 5 — Detect possible behavior patterns
Use evidence from journal notes and action flags to tag possible trading behavior patterns. Always tie a tag to evidence and use non-diagnostic language.
Supported MVP tags:
- `fomo_entry`
- `revenge_trade`
- `premature_exit`
- `overconfidence_after_winner`
- `stop_moved`
- `size_creep`
- `hesitation`
- `rule_drift`
- `no_pattern_detected`
### Step 6 — Produce next-session operating rules
Convert findings into temporary, concrete guardrails. Examples:
- require thesis record and screenshot before the next entry
- cap risk at 0.5R for the next two trades after a rule violation
- switch to review-only mode after repeated revenge-trade evidence
- do not chase a missed entry; add to watchlist for the next valid setup
### Step 7 — Human decision gate
End every report with a human decision gate. The default action is `journal_only`.
Allowed actions:
```text
accept_rules / modify_rules / defer / journal_only
```
## Output
The skill produces a JSON report and optionally a Markdown report.
Required top-level JSON fields:
- `schema_version`
- `review_type`
- `review_id`
- `overall_verdict`
- `summary`
- `scores`
- `process_adherence_findings`
- `risk_manager_notes`
- `execution_quality_assessment`
- `behavioral_pattern_tags`
- `next_session_operating_rules`
- `coach_questions`
- `human_decision_gate`
- `disclaimer`
Verdicts:
| Verdict | Meaning |
|---|---|
| `OK` | No material process violation found. Outcome appears compatible with the plan. |
| `WARN` | Minor process or record-quality concern. |
| `REVIEW_REQUIRED` | Meaningful process, risk, or behavior finding before next similar trade. |
| `RULE_VIOLATION` | Explicit user rule appears to have been broken. |
| `COOL_DOWN` | Repeated violations, drawdown/revenge pattern, or escalation suggests review-only mode. |
## Example Command
```bash
python3 skills/trade-performance-coach/scripts/review_trade_performance.py \
--input skills/trade-performance-coach/scripts/tests/fixtures/single_trade_rule_violation_loss.json \
--output-dir reports/trade-performance-coach \
--markdown
```
## Resources
Read these selectively when invoked:
- `references/review-framework.md` — five-axis review model, scoring, verdicts
- `references/behavior-tags.md` — behavior tag definitions and evidence rules
- `references/risk-review-checklist.md` — risk manager checklist and severity rules
- `references/output-contract.md` — JSON output contract and schema notes
- `references/hermes-integration.md` — suggested Hermes `/post-trade-coach` and monthly coaching integration
- `assets/performance_coach_report.schema.json` — machine-readable output schema
- `scripts/review_trade_performance.py` — deterministic local reviewer
## Guardrails
- This is process-review support, not financial advice.
- Do not recommend buying, selling, shorting, holding, or sizing a specific security.
- Do not provide therapy or mental-health diagnosis.
- Do not infer personality traits.
- Do not shame or moralize the user.
- Tie every behavior tag to evidence.
- Use "possible pattern" language for behavior tags.
- Always include a human decision gate.
- Default to journal/review mode when data is incomplete.