From dbcead60df3871e78ccc2c77d7fb013c7ffd895f Mon Sep 17 00:00:00 2001 From: agb1986 Date: Sat, 8 Aug 2026 12:41:09 +0100 Subject: [PATCH] feat: ground the trader in market data and harden unattended runs - daily-pipeline: fetch check_stock quote/history before each trader run and pass a condensed context file to the panel, so verdicts are based on real prices instead of training data - daily-pipeline: render the crypto portfolio in the report, send a best-effort failure alert email on stage errors, and make the run lock atomic - trader: enforce the judge's verdict JSON with structured outputs - financial-news: dedup syndicated articles by URL/title; keyword score is now a capped match count instead of a fraction of all configured keywords - common: retries with backoff and a User-Agent on the shared HTTP client - mcp: per-script timeout and shared logger in the MCP server - ci: GitHub Actions workflows for tests/lint and GHCR image publishing - docs: README sections for trader and daily-pipeline; deployment guide updated for alerting and CI Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 29 ++++ .github/workflows/docker.yml | 38 +++++ README.md | 55 +++++-- common/src/common/http_client.py | 22 +++ common/tests/test_http_client.py | 23 ++- docs/deployment.md | 19 ++- mcp_server/src/mcp_server/server.py | 26 +++- skills/daily_pipeline/SKILL.md | 25 ++-- skills/daily_pipeline/pipeline.yaml | 1 + skills/daily_pipeline/scripts/build_report.py | 2 + skills/daily_pipeline/scripts/run_daily.py | 25 +++- .../src/daily_pipeline/config.py | 2 +- .../src/daily_pipeline/context.py | 74 +++++++++ .../src/daily_pipeline/report.py | 48 ++++++ .../src/daily_pipeline/runner.py | 60 +++++++- .../daily_pipeline/tests/test_build_report.py | 60 ++++++++ skills/daily_pipeline/tests/test_context.py | 106 +++++++++++++ skills/daily_pipeline/tests/test_run_daily.py | 141 +++++++++++++++++- skills/financial_news/SKILL.md | 6 +- skills/financial_news/scripts/fetch_news.py | 5 +- .../src/financial_news/analyzer.py | 11 +- .../src/financial_news/fetcher.py | 27 ++++ .../financial_news/tests/test_analyze_news.py | 30 +++- .../financial_news/tests/test_fetch_news.py | 55 ++++++- skills/trader/src/trader/client.py | 24 ++- skills/trader/src/trader/judge.py | 64 +++++++- skills/trader/tests/test_run_judge.py | 22 ++- 27 files changed, 930 insertions(+), 70 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 .github/workflows/docker.yml create mode 100644 skills/daily_pipeline/src/daily_pipeline/context.py create mode 100644 skills/daily_pipeline/tests/test_context.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..e6e91ee --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,29 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Install uv + uses: astral-sh/setup-uv@v5 + with: + enable-cache: true + + - name: Sync workspace + run: uv sync --all-packages + + - name: Ruff lint + run: uv run ruff check . + + - name: Ruff format check + run: uv run ruff format --check . + + - name: Pytest (coverage threshold enforced by pytest config) + run: uv run pytest diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml new file mode 100644 index 0000000..c542be2 --- /dev/null +++ b/.github/workflows/docker.yml @@ -0,0 +1,38 @@ +# Builds the pipeline/MCP image on main and pushes it to GHCR, so the CasaOS +# host can `docker compose pull` instead of building ~1GB of deps locally. +name: Docker image + +on: + push: + branches: [main] + workflow_dispatch: + +jobs: + build-push: + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + steps: + - uses: actions/checkout@v4 + + - name: Log in to GHCR + uses: docker/login-action@v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Set up Buildx + uses: docker/setup-buildx-action@v3 + + - name: Build and push + uses: docker/build-push-action@v6 + with: + context: . + push: true + tags: | + ghcr.io/${{ github.repository }}:latest + ghcr.io/${{ github.repository }}:${{ github.sha }} + cache-from: type=gha + cache-to: type=gha,mode=max diff --git a/README.md b/README.md index ff5df2f..4a46e87 100644 --- a/README.md +++ b/README.md @@ -59,24 +59,27 @@ Fetches cryptocurrency quotes and OHLC history from CoinGecko, then generates AS ### `/financial-news` -Fetches the latest financial news from 10 RSS feeds (MarketWatch, Yahoo Finance, CNBC, Investing.com, Motley Fool) and ranks articles by keyword match and semantic similarity using a local sentence-transformer model. +Fetches the latest financial news from 10 RSS feeds (MarketWatch, Yahoo Finance, CNBC, Investing.com, Motley Fool), removes duplicates syndicated across feeds, and ranks articles by keyword match and semantic similarity using a local sentence-transformer model. **Usage** ``` /financial-news /financial-news --hours 6 +/financial-news --min-semantic 0.40 ``` -| Argument | Required | Default | Description | -|-------------|----------|---------|------------------------------------------| -| `--hours` | No | `24` | Look-back window in hours | -| `--keywords`| No | — | Path to a `keywords.json` config file | +| Argument | Required | Default | Description | +|------------------|----------|---------|------------------------------------------------| +| `--hours` | No | `24` | Look-back window in hours | +| `--keywords` | No | — | Path to a `keywords.json` config file | +| `--min-semantic` | No | — | Only include articles with semantic score ≥ N | +| `--min-keyword` | No | — | Only include articles with keyword score ≥ N | **What it returns** -- Fetched articles grouped by source, newest first -- Ranked list with keyword score (0–1) and semantic score (0–1) +- Fetched articles grouped by source, newest first, deduplicated by URL/title +- Ranked list with keyword score (0–1; capped count — 1 match = 0.33, 3+ = 1.0) and semantic score (0–1) - Group 1 — keyword match **and** semantic score ≥ 0.45 - Group 2 — semantic score ≥ 0.40 with no keyword match - An observation paragraph summarising score spread and relevance @@ -123,18 +126,40 @@ No arguments. Requires `CRYPTO_API_KEY` and `CRYPTO_API_SECRET` environment vari --- -### `/financial-news` — scoring filter +### `/trader` + +Runs a multi-agent investment analysis for a symbol: a five-role analyst panel (fundamental, technical, macro, sentiment, risk) in parallel, a fixed-rounds bull/bear debate over the panel's findings, and an impartial judge that scores the debate and returns a structured verdict. -The `analyze_news.py` script now accepts `--min-semantic` and `--min-keyword` flags to pre-filter articles before writing output, eliminating ad-hoc filtering steps. +**Usage** ``` -/financial-news --min-semantic 0.40 +/trader --symbol AMZN ``` -| Argument | Required | Default | Description | -|------------------|----------|---------|------------------------------------------------| -| `--min-semantic` | No | — | Only include articles with semantic score ≥ N | -| `--min-keyword` | No | — | Only include articles with keyword score ≥ N | +| Argument | Required | Default | Description | +|------------------|----------|---------|----------------------------------------------------| +| `--symbol` | Yes | | Ticker symbol to analyse | +| `--context-file` | No | — | Market data file to ground the panel (quote/history) | +| `--rounds` | No | `2` | Debate rounds | + +**What it returns** + +- Panel briefs, debate transcript, and a judge scorecard (evidence quality, rebuttal validity, risk-adjusted logic, internal consistency, falsifiability) with winner, confidence, and an actionable verdict +- The verdict JSON is schema-enforced via structured outputs + +> Each full run makes 10 Anthropic API calls (5× Sonnet panel, 4× Opus debate, 1× Opus judge), measured at ~$0.16. Models and prompts live in `skills/trader/config/roles.yaml`. + +--- + +### `/daily-pipeline` + +Orchestrates everything into one checkpointed daily run: fetch + rank news, extract top ticker mentions, fetch portfolios, ground the trader with `check_stock` market data, run it on the most relevant symbols, build a Markdown report (with token usage and estimated cost), and email it. Re-running the same day resumes from the first incomplete stage; on failure it sends a best-effort alert email. + +``` +uv run skills/daily_pipeline/scripts/run_daily.py [--dry-run] [--skip-email] [--no-summary] +``` + +Thresholds, caps, and pricing live in `skills/daily_pipeline/pipeline.yaml`. See `skills/daily_pipeline/SKILL.md` for the full stage list and `docs/deployment.md` for running it on a schedule. --- @@ -166,7 +191,7 @@ uv run python --version test -d .venv && echo "venv OK" # Key packages -uv run python -c "import check_stock, check_crypto, financial_news, get_stock_portfolio, get_crypto_portfolio, common" && echo "All packages OK" +uv run python -c "import check_stock, check_crypto, financial_news, get_stock_portfolio, get_crypto_portfolio, trader, daily_pipeline, mcp_server, common" && echo "All packages OK" ``` --- diff --git a/common/src/common/http_client.py b/common/src/common/http_client.py index 1f2a7ea..83958ed 100644 --- a/common/src/common/http_client.py +++ b/common/src/common/http_client.py @@ -1,9 +1,19 @@ """Generic HTTP client using requests.Session, shared across all skills.""" import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry from common.logger import get_logger +USER_AGENT = "finance-agent/0.1 (+https://github.com/agb1986/finance-agent)" + +# The pipeline runs unattended, so transient failures (CoinGecko 429s, feed +# hiccups, 5xx blips) must not fail a stage outright. Retry respects Retry-After. +DEFAULT_RETRIES = 3 +DEFAULT_BACKOFF = 1.0 +RETRY_STATUSES = (429, 500, 502, 503, 504) + class HttpClient: """Thin wrapper around requests.Session with a fixed base URL and shared headers.""" @@ -13,12 +23,24 @@ def __init__( base_url: str, headers: dict[str, str] | None = None, timeout: int = 30, + retries: int = DEFAULT_RETRIES, ) -> None: self.base_url = base_url.rstrip("/") self.timeout = timeout self.session = requests.Session() + self.session.headers["User-Agent"] = USER_AGENT if headers: self.session.headers.update(headers) + retry = Retry( + total=retries, + backoff_factor=DEFAULT_BACKOFF, + status_forcelist=RETRY_STATUSES, + allowed_methods=("GET", "POST"), + respect_retry_after_header=True, + ) + adapter = HTTPAdapter(max_retries=retry) + self.session.mount("https://", adapter) + self.session.mount("http://", adapter) def get(self, path: str, params: dict | None = None) -> dict | list: """Perform a GET request and return the parsed JSON response. diff --git a/common/tests/test_http_client.py b/common/tests/test_http_client.py index fc26a7c..3b66750 100644 --- a/common/tests/test_http_client.py +++ b/common/tests/test_http_client.py @@ -5,7 +5,7 @@ import pytest import requests -from common.http_client import HttpClient +from common.http_client import RETRY_STATUSES, USER_AGENT, HttpClient @pytest.fixture(autouse=True) @@ -35,6 +35,27 @@ def test_custom_timeout(self): client = HttpClient("https://example.com", timeout=10) assert client.timeout == 10 + def test_user_agent_set(self): + client = HttpClient("https://example.com") + assert client.session.headers["User-Agent"] == USER_AGENT + + def test_custom_headers_do_not_remove_user_agent(self): + client = HttpClient("https://example.com", headers={"Authorization": "key123"}) + assert client.session.headers["User-Agent"] == USER_AGENT + + def test_retry_adapter_mounted(self): + client = HttpClient("https://example.com") + for prefix in ("https://", "http://"): + retry = client.session.get_adapter(f"{prefix}example.com").max_retries + assert retry.total == 3 + assert set(RETRY_STATUSES) <= set(retry.status_forcelist) + assert "POST" in retry.allowed_methods + + def test_retry_count_configurable(self): + client = HttpClient("https://example.com", retries=5) + retry = client.session.get_adapter("https://example.com").max_retries + assert retry.total == 5 + class TestHttpClientGet: def _client(self, headers: dict | None = None) -> HttpClient: diff --git a/docs/deployment.md b/docs/deployment.md index 172d9ed..6789814 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -174,9 +174,18 @@ reported on a separate line so they never inflate the day's total. | `lock file ... another run in progress?` | A previous run is still going, or died hard. Locks older than 6h are cleared automatically. | | Model download attempted at runtime | `HF_HUB_OFFLINE=1` is set because the model is baked in. If you change the embedding model, rebuild the image. | -## Not yet done +## Failure alerting -- **Failure alerting** — a failed stage currently just exits non-zero into the - cron log; there is no push/email notification. -- **CI/CD** — no GitHub Actions job builds and pushes the image to GHCR yet, so - the CasaOS host builds locally. +When a required stage fails, the pipeline exits non-zero **and** sends a +best-effort alert email ("Daily Finance Report — pipeline FAILED") using the +same SMTP settings as the report. If SMTP itself is the failure, the alert is +skipped and the error stays in the cron log. + +## CI/CD + +GitHub Actions runs tests + ruff on every PR (`.github/workflows/ci.yml`) and +builds + pushes the image to GHCR on every merge to main +(`.github/workflows/docker.yml`). To consume the prebuilt image on the CasaOS +host instead of building locally, point the compose services at +`ghcr.io/agb1986/finance-agent:latest` and run `docker compose pull` — the +local `build:` fallback keeps working either way. diff --git a/mcp_server/src/mcp_server/server.py b/mcp_server/src/mcp_server/server.py index 4da4878..4619b96 100644 --- a/mcp_server/src/mcp_server/server.py +++ b/mcp_server/src/mcp_server/server.py @@ -5,12 +5,17 @@ import subprocess from pathlib import Path +from common.logger import get_logger, setup from mcp.server.fastmcp import FastMCP REPO_ROOT = Path(os.environ.get("REPO_ROOT", Path.cwd())) PORT = int(os.environ.get("MCP_PORT", "35001")) HOST = os.environ.get("MCP_HOST", "0.0.0.0") +# Mirrors the daily pipeline's per-script cap. Without a timeout, one hung +# yfinance call would wedge the tool call (and its SSE connection) forever. +SCRIPT_TIMEOUT = 900 + mcp = FastMCP("Finance Agent", host=HOST, port=PORT) @@ -21,14 +26,22 @@ def _run(script: str, args: list[str]) -> str: """Run a skill script via uv run and return stripped stdout.""" - result = subprocess.run( - ["uv", "run", script, *args], - capture_output=True, - text=True, - cwd=REPO_ROOT, - ) + logger = get_logger() + logger.debug(f"running: {script} {' '.join(args)}") + try: + result = subprocess.run( + ["uv", "run", script, *args], + capture_output=True, + text=True, + cwd=REPO_ROOT, + timeout=SCRIPT_TIMEOUT, + ) + except subprocess.TimeoutExpired as exc: + logger.error(f"{script} timed out after {SCRIPT_TIMEOUT}s") + raise RuntimeError(f"Script '{script}' timed out after {SCRIPT_TIMEOUT}s") from exc if result.returncode != 0: detail = result.stderr.strip() or f"exit code {result.returncode}" + logger.error(f"{script} failed: {detail}") raise RuntimeError(f"Script '{script}' failed: {detail}") return result.stdout.strip() @@ -156,4 +169,5 @@ def analyze_financial_news(hours: int = 24, min_semantic: float = 0.40) -> str: # --------------------------------------------------------------------------- if __name__ == "__main__": + setup(debug=os.environ.get("MCP_DEBUG", "") != "") mcp.run(transport="sse") diff --git a/skills/daily_pipeline/SKILL.md b/skills/daily_pipeline/SKILL.md index 0902872..9fcbce9 100644 --- a/skills/daily_pipeline/SKILL.md +++ b/skills/daily_pipeline/SKILL.md @@ -1,6 +1,6 @@ --- name: daily-pipeline -version: 0.1.2 +version: 0.1.3 description: Runs the full daily finance pipeline — fetches and ranks news, extracts top ticker mentions, fetches portfolios, runs the trader on the most relevant symbols, builds a Markdown report, and emails it. Use when the user asks for the daily report, the full pipeline, or a morning briefing. --- @@ -10,10 +10,16 @@ Orchestrates the other skills into one checkpointed daily run: ``` fetch_news → analyze_news → extract_tickers ─┐ -stock_portfolio → select_candidates ─────────┼→ trader (per symbol) → build_report → send_email -crypto_portfolio (artifact only, v1) ────────┘ +stock_portfolio → select_candidates ─────────┼→ market context → trader (per symbol) → build_report → send_email +crypto_portfolio (rendered in report) ───────┘ ``` +Before each trader run the pipeline fetches the symbol's quote and history via +`check_stock` (free Yahoo data) and passes a condensed context file to the +analyst panel, so verdicts are grounded in current prices rather than training +data. The lookback is `trader.history_period` in `pipeline.yaml`. If the fetch +fails the panel still runs, just ungrounded. + Each run writes to `tmp/run_/` with a `manifest.json` checkpoint — re-running the same day resumes from the first incomplete stage, so a crash or a cron re-fire never repeats completed API spend. Trader verdicts newer than @@ -108,8 +114,9 @@ skills/daily_pipeline/tmp/run_20260801/report.md ``` If SMTP is not configured, pass `--skip-email`. On stage failure the script -exits non-zero with the failing stage in the error log — re-running resumes -from that stage. +exits non-zero with the failing stage in the error log, and sends a best-effort +failure alert email using the same SMTP settings as the report — re-running +resumes from the failed stage. ### Step 3 — Present the result @@ -127,7 +134,7 @@ Each stage is also runnable standalone (all print their output path on stdout): ```bash uv run skills/daily_pipeline/scripts/extract_tickers.py --input [--top-n 5] uv run skills/daily_pipeline/scripts/select_candidates.py --input [--min-pnl-pct 5] -uv run skills/daily_pipeline/scripts/build_report.py --analysis --tickers [--portfolio ] [--verdict ]... +uv run skills/daily_pipeline/scripts/build_report.py --analysis --tickers [--portfolio ] [--crypto ] [--verdict ]... uv run skills/daily_pipeline/scripts/send_email.py --report [--subject "..."] ``` @@ -146,7 +153,5 @@ persist in named volumes rather than the working tree. ## Known limitations (v1) -- The crypto portfolio is fetched and recorded as an artifact but not yet - rendered in the report, and crypto holdings are not gated into trader runs. -- The trader panel runs without a market-data context file; wiring - `check_stock` quotes in as context is a planned improvement. +- Crypto holdings are rendered in the report but not yet gated into trader + runs — only stock positions feed `select_candidates`. diff --git a/skills/daily_pipeline/pipeline.yaml b/skills/daily_pipeline/pipeline.yaml index 9fc9d9c..21ee75d 100644 --- a/skills/daily_pipeline/pipeline.yaml +++ b/skills/daily_pipeline/pipeline.yaml @@ -16,6 +16,7 @@ trader: max_runs: 2 # hard cap on trader pipelines per day (10 API calls each) verdict_cache_days: 3 # reuse an existing verdict newer than this many days rounds: 2 # debate rounds passed to run_debate.py + history_period: "6mo" # check_stock lookback used to ground the panel with market data email: subject_prefix: "Daily Finance Report" diff --git a/skills/daily_pipeline/scripts/build_report.py b/skills/daily_pipeline/scripts/build_report.py index 9d1f590..b6468c1 100644 --- a/skills/daily_pipeline/scripts/build_report.py +++ b/skills/daily_pipeline/scripts/build_report.py @@ -31,6 +31,7 @@ def main() -> None: parser.add_argument("--analysis", type=Path, default=None, help="News analysis JSON") parser.add_argument("--tickers", type=Path, default=None, help="Extracted tickers JSON") parser.add_argument("--portfolio", type=Path, default=None, help="Stock portfolio JSON") + parser.add_argument("--crypto", type=Path, default=None, help="Crypto portfolio JSON") parser.add_argument( "--verdict", type=Path, @@ -50,6 +51,7 @@ def main() -> None: analysis_path=str(args.analysis) if args.analysis else None, tickers_path=str(args.tickers) if args.tickers else None, portfolio_path=str(args.portfolio) if args.portfolio else None, + crypto_path=str(args.crypto) if args.crypto else None, verdicts=[{"path": str(v), "cached": False} for v in args.verdict], stages={}, with_summary=not args.no_summary, diff --git a/skills/daily_pipeline/scripts/run_daily.py b/skills/daily_pipeline/scripts/run_daily.py index 228bf0a..efb9b03 100644 --- a/skills/daily_pipeline/scripts/run_daily.py +++ b/skills/daily_pipeline/scripts/run_daily.py @@ -17,10 +17,32 @@ from pathlib import Path from common.args import base_parser -from common.logger import setup +from common.logger import get_logger, setup from daily_pipeline.config import load_config +from daily_pipeline.emailer import send_report from daily_pipeline.runner import StageError, plan, run_pipeline +FAILURE_SUBJECT = "Daily Finance Report — pipeline FAILED" + + +def send_failure_alert(error: StageError) -> None: + """Best-effort failure notification reusing the report SMTP config. + + Without this, an unattended cron run fails silently into a log nobody + reads. Alert delivery must never mask the original failure, so any + error here is logged and swallowed. + """ + logger = get_logger() + body = ( + f"The daily finance pipeline failed:\n\n{error}\n\n" + "Re-running the pipeline resumes from the failed stage." + ) + try: + send_report(FAILURE_SUBJECT, body) + logger.debug("failure alert email sent") + except Exception as exc: + logger.error(f"failure alert email could not be sent: {exc}") + def main() -> None: parser = argparse.ArgumentParser( @@ -62,6 +84,7 @@ def main() -> None: ) except StageError as exc: logger.error(f"pipeline failed: {exc}") + send_failure_alert(exc) sys.exit(1) print(manifest["report"]) diff --git a/skills/daily_pipeline/src/daily_pipeline/config.py b/skills/daily_pipeline/src/daily_pipeline/config.py index 7ec26ac..7150921 100644 --- a/skills/daily_pipeline/src/daily_pipeline/config.py +++ b/skills/daily_pipeline/src/daily_pipeline/config.py @@ -13,7 +13,7 @@ "news": {"hours": 24, "min_semantic": 0.40}, "tickers": {"top_n": 5}, "portfolio": {"min_abs_pnl_pct": 5.0, "min_position_value": 25.0, "max_candidates": 3}, - "trader": {"max_runs": 2, "verdict_cache_days": 3, "rounds": 2}, + "trader": {"max_runs": 2, "verdict_cache_days": 3, "rounds": 2, "history_period": "6mo"}, "email": {"subject_prefix": "Daily Finance Report"}, "retention_days": 14, "pricing": { diff --git a/skills/daily_pipeline/src/daily_pipeline/context.py b/skills/daily_pipeline/src/daily_pipeline/context.py new file mode 100644 index 0000000..143ef91 --- /dev/null +++ b/skills/daily_pipeline/src/daily_pipeline/context.py @@ -0,0 +1,74 @@ +"""Build a compact market-data context for the trader panel. + +The raw check_stock history JSON runs to tens of kilobytes; sent to all five +panel roles it would multiply prompt spend for little analytical gain. This +condenses quote + history into the numbers the analysts actually use: current +quote, period range, simple moving averages, and the most recent bars. +""" + +from common.logger import get_logger + +RECENT_BARS = 10 +SMA_WINDOWS = (20, 50, 200) + + +def sma(closes: list[float], window: int) -> float | None: + """Simple moving average over the last `window` closes, or None if too few.""" + if len(closes) < window: + return None + return round(sum(closes[-window:]) / window, 4) + + +def _fmt(value: object, suffix: str = "") -> str: + if value is None: + return "n/a" + if isinstance(value, float): + return f"{value:,.2f}{suffix}" + if isinstance(value, int): + return f"{value:,}{suffix}" + return f"{value}{suffix}" + + +def build_context(quote: dict, history: dict) -> str: + """Render quote + condensed history as plain text for the panel prompt.""" + logger = get_logger() + symbol = quote.get("symbol", "?") + bars = history.get("bars", []) + closes = [bar["close"] for bar in bars] + + lines = [ + f"# {symbol} market data — fetched {quote.get('fetched_at', 'n/a')}", + f"Name: {quote.get('name') or 'n/a'}", + f"Price: {_fmt(quote.get('price'))} {quote.get('currency') or ''}".rstrip(), + ( + f"Change vs previous close: {_fmt(quote.get('change'))} " + f"({_fmt(quote.get('change_percent'), '%')})" + ), + f"Volume: {_fmt(quote.get('volume'))}", + f"Market cap: {_fmt(quote.get('market_cap'))}", + f"P/E ratio: {_fmt(quote.get('pe_ratio'))}", + f"52-week range: {_fmt(quote.get('week_52_low'))} – {_fmt(quote.get('week_52_high'))}", + "", + f"## Price history ({history.get('period', '?')}, daily, {len(bars)} bars)", + ] + + if bars: + lines.append( + f"Period range: low {_fmt(min(bar['low'] for bar in bars))} – " + f"high {_fmt(max(bar['high'] for bar in bars))}" + ) + sma_parts = [f"SMA{window}: {_fmt(sma(closes, window))}" for window in SMA_WINDOWS] + lines.append(" ".join(sma_parts)) + lines.append("") + lines.append(f"Last {min(RECENT_BARS, len(bars))} bars (oldest first):") + lines.append("date | open | high | low | close | volume") + for bar in bars[-RECENT_BARS:]: + lines.append( + f"{bar['date']} | {bar['open']} | {bar['high']} | {bar['low']} | " + f"{bar['close']} | {bar['volume']}" + ) + else: + lines.append("No history bars available.") + + logger.debug(f"build_context: {symbol} context built from {len(bars)} bars") + return "\n".join(lines) diff --git a/skills/daily_pipeline/src/daily_pipeline/report.py b/skills/daily_pipeline/src/daily_pipeline/report.py index ace2a3f..8d40b86 100644 --- a/skills/daily_pipeline/src/daily_pipeline/report.py +++ b/skills/daily_pipeline/src/daily_pipeline/report.py @@ -97,6 +97,47 @@ def format_portfolio_section(portfolio: dict | None) -> str: return "\n".join(lines) +def format_crypto_section(portfolio: dict | None) -> str: + """Render the Crypto.com balance and positions as Markdown.""" + if not portfolio: + return "_Crypto portfolio fetch was skipped or failed (missing credentials?)._" + lines: list[str] = [] + for balance in portfolio.get("balances", []): + lines.extend( + [ + f"**Currency:** {balance.get('instrument_name', '?')}", + f"**Cash balance:** {_as_float(balance.get('total_cash_balance')):.2f}", + f"**Available:** {_as_float(balance.get('total_available_balance')):.2f}", + ( + "**Session unrealised PnL:** " + f"{_as_float(balance.get('total_session_unrealized_pnl')):+.2f}" + ), + "", + ] + ) + positions = portfolio.get("position_balances", []) + if positions: + lines.append("| Instrument | Quantity | Market value |") + lines.append("|---|---|---|") + for position in sorted(positions, key=lambda p: -_as_float(p.get("market_value"))): + lines.append( + f"| {position.get('instrument_name', '?')} " + f"| {position.get('quantity', '0')} " + f"| {_as_float(position.get('market_value')):.2f} |" + ) + else: + lines.append("_No open crypto positions._") + return "\n".join(lines).rstrip() + + +def _as_float(value: object) -> float: + """Crypto.com reports numbers as strings; coerce defensively.""" + try: + return float(value or 0) + except (TypeError, ValueError): + return 0.0 + + def format_verdict_section(verdict_file: dict, cached: bool) -> str: """Render one trader verdict as the SKILL.md scorecard format.""" symbol = verdict_file.get("symbol", "?") @@ -184,6 +225,7 @@ def build_report( stages: dict, with_summary: bool = True, pricing: dict | None = None, + crypto_path: str | None = None, ) -> str: """Assemble the full Markdown report. @@ -196,6 +238,7 @@ def build_report( stages: Manifest stage dict for the run-status footer. with_summary: Set False to skip the Claude executive-summary call. pricing: The ``pricing`` config block used to cost the token usage. + crypto_path: Path to the crypto portfolio JSON, or None. Returns: The complete report as a Markdown string. @@ -204,6 +247,7 @@ def build_report( articles = _load_json(analysis_path) if analysis_path else [] tickers = _load_json(tickers_path) if tickers_path else [] portfolio = _load_json(portfolio_path) if portfolio_path else None + crypto = _load_json(crypto_path) if crypto_path else None sections = [ f"# Daily Finance Report — {date}", @@ -220,6 +264,10 @@ def build_report( "", format_portfolio_section(portfolio), "", + "## Crypto portfolio", + "", + format_crypto_section(crypto), + "", "## Trader verdicts", "", ] diff --git a/skills/daily_pipeline/src/daily_pipeline/runner.py b/skills/daily_pipeline/src/daily_pipeline/runner.py index 072059a..53f5a05 100644 --- a/skills/daily_pipeline/src/daily_pipeline/runner.py +++ b/skills/daily_pipeline/src/daily_pipeline/runner.py @@ -19,6 +19,7 @@ from common.logger import get_logger from daily_pipeline import candidates as candidates_mod +from daily_pipeline import context as context_mod from daily_pipeline import report as report_mod from daily_pipeline import tickers as tickers_mod from daily_pipeline.emailer import send_report @@ -33,6 +34,8 @@ "analyze_news": "skills/financial_news/scripts/analyze_news.py", "stock_portfolio": "skills/get_stock_portfolio/scripts/fetch_portfolio.py", "crypto_portfolio": "skills/get_crypto_portfolio/scripts/fetch_portfolio.py", + "fetch_quote": "skills/check_stock/scripts/fetch_quote.py", + "fetch_history": "skills/check_stock/scripts/fetch_history.py", "run_panel": "skills/trader/scripts/run_panel.py", "run_debate": "skills/trader/scripts/run_debate.py", "run_judge": "skills/trader/scripts/run_judge.py", @@ -118,7 +121,7 @@ def record_stage( def acquire_lock(run_dir: Path) -> Path: - """Create a lock file; refuse to start when a fresh lock already exists. + """Atomically create a lock file; refuse to start when a fresh lock exists. Raises: StageError: If another run appears to be in progress. @@ -130,7 +133,12 @@ def acquire_lock(run_dir: Path) -> Path: raise StageError(f"lock file {lock_path} is {age:.0f}s old — another run in progress?") get_logger().debug(f"removing stale lock ({age:.0f}s old)") lock_path.unlink() - lock_path.write_text(str(os.getpid())) + try: + # open("x") is atomic: two simultaneous cron fires cannot both win. + with lock_path.open("x") as f: + f.write(str(os.getpid())) + except FileExistsError as exc: + raise StageError(f"lock file {lock_path} appeared concurrently — another run won") from exc return lock_path @@ -175,9 +183,38 @@ def find_cached_verdict( return newest[1] if newest else None -def run_trader(symbol: str, rounds: int) -> str: +def build_market_context(symbol: str, run_dir: Path, period: str) -> Path | None: + """Fetch quote + history via check_stock and write a compact context file. + + Returns None when the fetch fails — the panel still runs, just ungrounded, + matching the trader's standalone behaviour. Grounding the panel in real + market data is what keeps the technical analyst's price levels honest. + """ + logger = get_logger() + try: + quote_path = run_script(SCRIPTS["fetch_quote"], ["--symbol", symbol]) + history_path = run_script( + SCRIPTS["fetch_history"], ["--symbol", symbol, "--period", period] + ) + except StageError as exc: + logger.error(f"{symbol}: market data fetch failed ({exc}) — panel runs without context") + return None + with Path(quote_path).open() as f: + quote = json.load(f) + with Path(history_path).open() as f: + history = json.load(f) + context_path = run_dir / f"context_{symbol}.md" + context_path.write_text(context_mod.build_context(quote, history)) + logger.debug(f"{symbol}: market context written to {context_path}") + return context_path + + +def run_trader(symbol: str, rounds: int, context_file: str | None = None) -> str: """Run the three trader stages for a symbol and return the verdict path.""" - panel_path = run_script(SCRIPTS["run_panel"], ["--symbol", symbol]) + panel_args = ["--symbol", symbol] + if context_file: + panel_args += ["--context-file", context_file] + panel_path = run_script(SCRIPTS["run_panel"], panel_args) debate_path = run_script( SCRIPTS["run_debate"], ["--input", panel_path, "--rounds", str(rounds)] ) @@ -201,7 +238,8 @@ def plan(config: dict) -> list[str]: ), ( f"trader (max {config['trader']['max_runs']} runs, " - f"verdict cache {config['trader']['verdict_cache_days']}d)" + f"verdict cache {config['trader']['verdict_cache_days']}d, " + f"grounded on {config['trader']['history_period']} market data)" ), "build_report", "send_email (skipped unless SMTP_* configured)", @@ -331,12 +369,18 @@ def run_pipeline( } save_manifest(run_dir, manifest) continue + context_path = build_market_context(symbol, run_dir, config["trader"]["history_period"]) try: - verdict_path = run_trader(symbol, config["trader"]["rounds"]) + verdict_path = run_trader( + symbol, + config["trader"]["rounds"], + context_file=str(context_path) if context_path else None, + ) manifest["trader"][symbol] = { "status": "done", "verdict": verdict_path, "cached": False, + "context": str(context_path) if context_path else None, } except StageError as exc: logger.error(f"trader failed for {symbol}: {exc}") @@ -349,6 +393,9 @@ def run_pipeline( for entry in manifest["trader"].values() if entry.get("status") == "done" ] + crypto_path = manifest["stages"].get("crypto_portfolio", {}).get("output") + if crypto_path and not Path(crypto_path).exists(): + crypto_path = None report_body = report_mod.build_report( date=f"{date[:4]}-{date[4:6]}-{date[6:]}", analysis_path=analysis_path, @@ -358,6 +405,7 @@ def run_pipeline( stages=manifest["stages"], with_summary=with_summary, pricing=config.get("pricing"), + crypto_path=crypto_path, ) report_path = run_dir / "report.md" report_path.write_text(report_body) diff --git a/skills/daily_pipeline/tests/test_build_report.py b/skills/daily_pipeline/tests/test_build_report.py index 1c5d054..c005da4 100644 --- a/skills/daily_pipeline/tests/test_build_report.py +++ b/skills/daily_pipeline/tests/test_build_report.py @@ -81,6 +81,23 @@ def reset_logger(): } +CRYPTO = { + "fetched_at": "2026-08-08T07:00:00Z", + "balances": [ + { + "instrument_name": "USD", + "total_cash_balance": "1250.50", + "total_available_balance": "1000.00", + "total_session_unrealized_pnl": "-12.34", + } + ], + "position_balances": [ + {"instrument_name": "BTC", "quantity": "0.015", "market_value": "900.00"}, + {"instrument_name": "SOL", "quantity": "5", "market_value": "350.50"}, + ], +} + + # ── section formatters ──────────────────────────────────────────────────────── @@ -110,6 +127,32 @@ def test_portfolio_section_renders_positions(self): def test_portfolio_section_none(self): assert "skipped or failed" in report_mod.format_portfolio_section(None) + def test_crypto_section_renders_balance_and_positions(self): + result = report_mod.format_crypto_section(CRYPTO) + assert "**Currency:** USD" in result + assert "**Cash balance:** 1250.50" in result + assert "**Session unrealised PnL:** -12.34" in result + assert "| BTC | 0.015 | 900.00 |" in result + + def test_crypto_section_positions_sorted_by_value(self): + result = report_mod.format_crypto_section(CRYPTO) + assert result.index("BTC") < result.index("SOL") + + def test_crypto_section_none(self): + assert "skipped or failed" in report_mod.format_crypto_section(None) + + def test_crypto_section_no_positions(self): + crypto = {**CRYPTO, "position_balances": []} + assert "No open crypto positions" in report_mod.format_crypto_section(crypto) + + def test_crypto_section_bad_numbers_coerced(self): + crypto = { + "balances": [{"instrument_name": "USD", "total_cash_balance": "not-a-number"}], + "position_balances": [], + } + result = report_mod.format_crypto_section(crypto) + assert "**Cash balance:** 0.00" in result + def test_verdict_section_scorecard(self): result = report_mod.format_verdict_section(VERDICT_FILE, cached=False) assert "### AMZN — 2026-08-01" in result @@ -199,6 +242,23 @@ def test_assembles_all_sections(self, tmp_path): assert "### AMZN" in body assert "- fetch_news: done" in body + def test_crypto_section_included_when_path_given(self, tmp_path): + analysis, tickers, _ = self._fixture_paths(tmp_path) + crypto = tmp_path / "crypto.json" + crypto.write_text(json.dumps(CRYPTO)) + body = report_mod.build_report( + date="2026-08-01", + analysis_path=str(analysis), + tickers_path=str(tickers), + portfolio_path=None, + crypto_path=str(crypto), + verdicts=[], + stages={}, + with_summary=False, + ) + assert "## Crypto portfolio" in body + assert "| BTC | 0.015 | 900.00 |" in body + def test_no_verdicts_placeholder(self, tmp_path): analysis, tickers, _ = self._fixture_paths(tmp_path) body = report_mod.build_report( diff --git a/skills/daily_pipeline/tests/test_context.py b/skills/daily_pipeline/tests/test_context.py new file mode 100644 index 0000000..5caf32f --- /dev/null +++ b/skills/daily_pipeline/tests/test_context.py @@ -0,0 +1,106 @@ +"""Tests for daily_pipeline.context — the trader market-data context builder.""" + +import logging + +import pytest +from daily_pipeline.context import build_context, sma + + +@pytest.fixture(autouse=True) +def reset_logger(): + yield + logging.getLogger("finance_agent").handlers.clear() + + +def _quote(**overrides) -> dict: + quote = { + "symbol": "AMZN", + "fetched_at": "2026-08-08T07:00:00Z", + "name": "Amazon.com, Inc.", + "price": 185.5, + "currency": "USD", + "change": 1.2, + "change_percent": 0.65, + "volume": 12_345_678, + "market_cap": 1_900_000_000_000, + "pe_ratio": 40.5, + "week_52_high": 200.0, + "week_52_low": 140.0, + } + quote.update(overrides) + return quote + + +def _history(n_bars: int = 30) -> dict: + return { + "symbol": "AMZN", + "period": "6mo", + "fetched_at": "2026-08-08T07:00:00Z", + "bars": [ + { + "date": f"2026-06-{(i % 28) + 1:02d}", + "open": 100.0 + i, + "high": 101.0 + i, + "low": 99.0 + i, + "close": 100.5 + i, + "volume": 1000 + i, + } + for i in range(n_bars) + ], + } + + +class TestSma: + def test_average_of_last_window(self): + assert sma([1.0, 2.0, 3.0, 4.0], 2) == 3.5 + + def test_none_when_too_few_closes(self): + assert sma([1.0, 2.0], 5) is None + + def test_exact_window(self): + assert sma([2.0, 4.0], 2) == 3.0 + + +class TestBuildContext: + def test_contains_quote_fields(self): + text = build_context(_quote(), _history()) + assert "# AMZN market data" in text + assert "Amazon.com, Inc." in text + assert "185.50 USD" in text + assert "P/E ratio: 40.50" in text + assert "52-week range: 140.00 – 200.00" in text + + def test_contains_history_summary(self): + text = build_context(_quote(), _history(30)) + assert "## Price history (6mo, daily, 30 bars)" in text + assert "SMA20:" in text + assert "Period range:" in text + + def test_sma_na_when_insufficient_bars(self): + text = build_context(_quote(), _history(10)) + assert "SMA20: n/a" in text + assert "SMA200: n/a" in text + + def test_recent_bars_capped_at_ten(self): + text = build_context(_quote(), _history(30)) + assert "Last 10 bars (oldest first):" in text + # 10 bar rows plus the header row use the pipe separator + bar_rows = [line for line in text.splitlines() if line.count(" | ") == 5] + assert len(bar_rows) == 11 + + def test_missing_quote_fields_render_na(self): + quote = _quote(price=None, pe_ratio=None, name=None) + text = build_context(quote, _history()) + assert "Price: n/a" in text + assert "P/E ratio: n/a" in text + assert "Name: n/a" in text + + def test_empty_history_handled(self): + text = build_context(_quote(), {"period": "6mo", "bars": []}) + assert "No history bars available." in text + + def test_stays_compact(self): + # The whole point: the context must stay small enough to send to five + # panel roles without materially increasing prompt spend. + text = build_context(_quote(), _history(180)) + assert len(text) < 2500 diff --git a/skills/daily_pipeline/tests/test_run_daily.py b/skills/daily_pipeline/tests/test_run_daily.py index 00e3b8d..fc3f714 100644 --- a/skills/daily_pipeline/tests/test_run_daily.py +++ b/skills/daily_pipeline/tests/test_run_daily.py @@ -18,6 +18,7 @@ from daily_pipeline.runner import ( StageError, acquire_lock, + build_market_context, find_cached_verdict, load_manifest, plan, @@ -146,6 +147,13 @@ def test_stale_lock_replaced(self, tmp_path): os.utime(lock, (stale, stale)) assert acquire_lock(tmp_path).exists() + def test_concurrent_lock_creation_raises(self, tmp_path): + # Simulate the loser of the exists-check race: the file appears between + # the check and the atomic create. + with patch.object(Path, "open", side_effect=FileExistsError): + with pytest.raises(StageError, match="another run won"): + acquire_lock(tmp_path) + def test_prune_removes_only_old_runs(self, tmp_path): (tmp_path / "run_20260701").mkdir() (tmp_path / "run_20260731").mkdir() @@ -184,8 +192,79 @@ def test_run_trader_chains_stages(self): ) as run_mock: assert run_trader("AMZN", rounds=2) == "verdict.json" assert run_mock.call_count == 3 + assert run_mock.call_args_list[0].args[1] == ["--symbol", "AMZN"] assert run_mock.call_args_list[1].args[1] == ["--input", "panel.json", "--rounds", "2"] + def test_run_trader_passes_context_file_to_panel(self): + with patch.object( + runner, "run_script", side_effect=["panel.json", "debate.json", "verdict.json"] + ) as run_mock: + run_trader("AMZN", rounds=2, context_file="ctx.md") + assert run_mock.call_args_list[0].args[1] == [ + "--symbol", + "AMZN", + "--context-file", + "ctx.md", + ] + + +# ── market context ──────────────────────────────────────────────────────────── + +QUOTE = { + "symbol": "AMZN", + "fetched_at": "2026-08-08T07:00:00Z", + "name": "Amazon.com, Inc.", + "price": 185.5, + "currency": "USD", + "change": 1.2, + "change_percent": 0.65, + "volume": 1000, + "market_cap": 2_000_000, + "pe_ratio": 40.5, + "week_52_high": 200.0, + "week_52_low": 140.0, +} + +HISTORY = { + "symbol": "AMZN", + "period": "6mo", + "fetched_at": "2026-08-08T07:00:00Z", + "bars": [ + { + "date": f"2026-07-{day:02d}", + "open": 180.0 + day, + "high": 181.0 + day, + "low": 179.0 + day, + "close": 180.5 + day, + "volume": 100 + day, + } + for day in range(1, 26) + ], +} + + +class TestBuildMarketContext: + def _paths(self, tmp_path): + quote_path = tmp_path / "quote.json" + quote_path.write_text(json.dumps(QUOTE)) + history_path = tmp_path / "history.json" + history_path.write_text(json.dumps(HISTORY)) + return str(quote_path), str(history_path) + + def test_writes_context_file(self, tmp_path): + quote_path, history_path = self._paths(tmp_path) + with patch.object(runner, "run_script", side_effect=[quote_path, history_path]) as run_mock: + result = build_market_context("AMZN", tmp_path, "6mo") + assert result == tmp_path / "context_AMZN.md" + content = result.read_text() + assert "AMZN" in content + assert "185.50" in content + assert run_mock.call_args_list[1].args[1] == ["--symbol", "AMZN", "--period", "6mo"] + + def test_fetch_failure_returns_none(self, tmp_path): + with patch.object(runner, "run_script", side_effect=StageError("yahoo down")): + assert build_market_context("AMZN", tmp_path, "6mo") is None + # ── plan ────────────────────────────────────────────────────────────────────── @@ -228,10 +307,16 @@ def pipeline_env(tmp_path, monkeypatch): verdict_path.write_text( json.dumps({"symbol": "AMZN", "generated_at": "2026-08-01T09:00:00Z", "verdict": {}}) ) + quote_path = tmp_path / "quote.json" + quote_path.write_text(json.dumps(QUOTE)) + history_path = tmp_path / "history.json" + history_path.write_text(json.dumps(HISTORY)) outputs = { "skills/financial_news/scripts/fetch_news.py": str(news_path), "skills/financial_news/scripts/analyze_news.py": str(analysis_path), + "skills/check_stock/scripts/fetch_quote.py": str(quote_path), + "skills/check_stock/scripts/fetch_history.py": str(history_path), } def fake_run_script(script, args=None, timeout=None): @@ -285,6 +370,34 @@ def counting_run_script(script, args=None, timeout=None): assert first_count == 2 assert len(calls) == 2 # nothing re-ran on resume + def test_trader_grounded_with_market_context(self, pipeline_env): + with ( + patch.object(runner, "find_cached_verdict", return_value=None), + patch.object( + runner, "run_trader", return_value=str(pipeline_env["verdict"]) + ) as trader_mock, + ): + manifest = run_pipeline(DEFAULTS, date="20260801", skip_email=True, with_summary=False) + + context_file = trader_mock.call_args.kwargs["context_file"] + assert context_file is not None + assert Path(context_file).exists() + assert "185.50" in Path(context_file).read_text() + assert manifest["trader"]["AMZN"]["context"] == context_file + + def test_trader_runs_ungrounded_when_market_data_fails(self, pipeline_env): + with ( + patch.object(runner, "find_cached_verdict", return_value=None), + patch.object(runner, "build_market_context", return_value=None), + patch.object( + runner, "run_trader", return_value=str(pipeline_env["verdict"]) + ) as trader_mock, + ): + manifest = run_pipeline(DEFAULTS, date="20260801", skip_email=True, with_summary=False) + + assert trader_mock.call_args.kwargs["context_file"] is None + assert manifest["trader"]["AMZN"]["status"] == "done" + def test_email_sent_when_configured(self, pipeline_env): with ( patch.object(runner, "find_cached_verdict", return_value=None), @@ -347,11 +460,37 @@ def test_success_prints_report_path(self, capsys): assert "report.md" in capsys.readouterr().out assert run_mock.call_args.kwargs["skip_email"] is True - def test_stage_error_exits_nonzero(self): + def test_stage_error_exits_nonzero_and_sends_alert(self): with ( patch.object(run_daily_script, "run_pipeline", side_effect=StageError("boom")), + patch.object(run_daily_script, "send_report") as send_mock, patch.object(sys, "argv", ["run_daily.py"]), pytest.raises(SystemExit) as excinfo, ): run_daily_script.main() assert excinfo.value.code == 1 + subject, body = send_mock.call_args.args + assert "FAILED" in subject + assert "boom" in body + + def test_alert_failure_does_not_mask_exit_code(self): + with ( + patch.object(run_daily_script, "run_pipeline", side_effect=StageError("boom")), + patch.object( + run_daily_script, "send_report", side_effect=RuntimeError("smtp not configured") + ), + patch.object(sys, "argv", ["run_daily.py"]), + pytest.raises(SystemExit) as excinfo, + ): + run_daily_script.main() + assert excinfo.value.code == 1 + + def test_success_sends_no_alert(self, capsys): + manifest = {"report": "tmp/run_20260801/report.md"} + with ( + patch.object(run_daily_script, "run_pipeline", return_value=manifest), + patch.object(run_daily_script, "send_report") as send_mock, + patch.object(sys, "argv", ["run_daily.py", "--skip-email"]), + ): + run_daily_script.main() + send_mock.assert_not_called() diff --git a/skills/financial_news/SKILL.md b/skills/financial_news/SKILL.md index b1c5bfa..33eaa5b 100644 --- a/skills/financial_news/SKILL.md +++ b/skills/financial_news/SKILL.md @@ -1,6 +1,6 @@ --- name: financial-news -version: 0.2.0 +version: 0.2.1 description: Fetches and summarises financial news from RSS feeds. Use when the user asks about today's market news, recent market movements, crypto news, earnings reports, or stock picks. --- @@ -51,7 +51,7 @@ skills/financial_news/tmp/news_results_20260315_143000.json ### Step 2 — Read the output file -Read the file path returned in Step 1. It contains a JSON array of articles sorted newest-first: +Read the file path returned in Step 1. It contains a JSON array of articles sorted newest-first, with duplicates (same URL or same title syndicated across feeds) already removed: ```json [ @@ -124,7 +124,7 @@ Read the file path returned in Step 2. It contains the same article array as the | Field | Description | |---|---| -| `keyword_score` | Fraction of configured keywords found in the article text (0.0–1.0) | +| `keyword_score` | Distinct keyword matches capped at 3, normalised to 0.0–1.0 (1 match = 0.33, 2 = 0.67, 3+ = 1.0) | | `semantic_score` | Cosine similarity between the article and keyword query using `all-MiniLM-L6-v2` (0.0–1.0) | | `matched_keywords` | List of keywords that were literally matched | | `rank` | Position after sorting by `semantic_score` descending | diff --git a/skills/financial_news/scripts/fetch_news.py b/skills/financial_news/scripts/fetch_news.py index d5eb4b0..f56149d 100644 --- a/skills/financial_news/scripts/fetch_news.py +++ b/skills/financial_news/scripts/fetch_news.py @@ -16,7 +16,7 @@ from common.args import base_parser from common.logger import setup -from financial_news.fetcher import FEEDS, fetch_feed +from financial_news.fetcher import FEEDS, dedup_articles, fetch_feed TMP_DIR = Path(__file__).parent.parent / "tmp" @@ -39,8 +39,9 @@ def main() -> None: articles = fetch_feed(name, url, cutoff) all_articles.extend(articles) + all_articles = dedup_articles(all_articles) all_articles.sort(key=lambda a: a["published_at"], reverse=True) - logger.debug(f"total articles after sort: {len(all_articles)}") + logger.debug(f"total articles after dedup + sort: {len(all_articles)}") TMP_DIR.mkdir(exist_ok=True) timestamp = time.strftime("%Y%m%d_%H%M%S") diff --git a/skills/financial_news/src/financial_news/analyzer.py b/skills/financial_news/src/financial_news/analyzer.py index 7992d7c..dc61b18 100644 --- a/skills/financial_news/src/financial_news/analyzer.py +++ b/skills/financial_news/src/financial_news/analyzer.py @@ -13,6 +13,12 @@ MODEL_NAME = "all-MiniLM-L6-v2" +# Matching this many keywords scores 1.0. A capped count keeps scores meaningful +# regardless of how many keywords are configured — with the old +# fraction-of-all-keywords formula, one hit out of 12 keywords scored 0.083 and +# thresholds like --min-keyword 0.5 were effectively unreachable. +KEYWORD_SCORE_CAP = 3 + def find_latest_news(tmp_dir: Path) -> Path: """Return the most recently created news_results JSON in tmp_dir.""" @@ -35,11 +41,12 @@ def load_keywords(config_path: Path) -> list[str]: def compute_keyword_score(text: str, keywords: list[str]) -> tuple[float, list[str]]: """Return (score, matched_keywords) for the given text against the keyword list. - Score is the fraction of keywords found in the text (0.0–1.0). + Score is the number of distinct keywords found, capped at KEYWORD_SCORE_CAP + and normalised to 0.0–1.0 (1 match = 1/3, 2 = 2/3, 3+ = 1.0). """ text_lower = text.lower() matched = [kw for kw in keywords if kw in text_lower] - score = len(matched) / len(keywords) + score = min(len(matched), KEYWORD_SCORE_CAP) / KEYWORD_SCORE_CAP return round(score, 4), matched diff --git a/skills/financial_news/src/financial_news/fetcher.py b/skills/financial_news/src/financial_news/fetcher.py index 4733795..019112b 100644 --- a/skills/financial_news/src/financial_news/fetcher.py +++ b/skills/financial_news/src/financial_news/fetcher.py @@ -4,6 +4,7 @@ Provides feed definitions and article fetching from RSS sources. """ +import re import time from html.parser import HTMLParser @@ -72,3 +73,29 @@ def fetch_feed(name: str, url: str, since: time.struct_time) -> list[dict]: logger.debug(f"{name}: {len(articles)} articles matched") return articles + + +def dedup_articles(articles: list[dict]) -> list[dict]: + """Drop articles that appear in more than one feed. + + The same story is frequently syndicated across feeds (six of the feeds are + Investing.com), and each copy would count as a separate ticker mention + downstream. Match on URL first, then on normalised title; keep the first + occurrence in fetch order. + """ + logger = get_logger() + seen_urls: set[str] = set() + seen_titles: set[str] = set() + unique: list[dict] = [] + for article in articles: + url = (article.get("url") or "").strip().rstrip("/").lower() + title = re.sub(r"\W+", " ", (article.get("title") or "").lower()).strip() + if (url and url in seen_urls) or (title and title in seen_titles): + continue + if url: + seen_urls.add(url) + if title: + seen_titles.add(title) + unique.append(article) + logger.debug(f"deduped {len(articles) - len(unique)} of {len(articles)} articles") + return unique diff --git a/skills/financial_news/tests/test_analyze_news.py b/skills/financial_news/tests/test_analyze_news.py index 88e1137..f5f4292 100644 --- a/skills/financial_news/tests/test_analyze_news.py +++ b/skills/financial_news/tests/test_analyze_news.py @@ -103,24 +103,44 @@ def test_ignores_blank_entries(self, tmp_path): class TestComputeKeywordScore: - def test_all_keywords_matched(self): - score, matched = compute_keyword_score("bitcoin etf rally", ["bitcoin", "etf"]) + def test_matches_at_cap_score_one(self): + score, matched = compute_keyword_score( + "bitcoin etf fed rally", ["bitcoin", "etf", "fed", "gold"] + ) assert score == 1.0 - assert set(matched) == {"bitcoin", "etf"} + assert set(matched) == {"bitcoin", "etf", "fed"} def test_no_keywords_matched(self): score, matched = compute_keyword_score("apple earnings beat", ["bitcoin", "etf"]) assert score == 0.0 assert matched == [] - def test_partial_match(self): + def test_single_match_is_one_third(self): score, matched = compute_keyword_score("bitcoin rally", ["bitcoin", "etf", "fed"]) assert score == pytest.approx(1 / 3, abs=0.001) assert matched == ["bitcoin"] + def test_two_matches_is_two_thirds(self): + score, matched = compute_keyword_score("bitcoin etf rally", ["bitcoin", "etf", "fed"]) + assert score == pytest.approx(2 / 3, abs=0.001) + assert set(matched) == {"bitcoin", "etf"} + + def test_score_independent_of_keyword_list_size(self): + many = [f"kw{i}" for i in range(10)] + ["bitcoin"] + score_many, _ = compute_keyword_score("bitcoin rally", many) + score_few, _ = compute_keyword_score("bitcoin rally", ["bitcoin", "etf"]) + assert score_many == score_few + + def test_matches_beyond_cap_still_one(self): + keywords = ["bitcoin", "etf", "fed", "gold", "oil"] + score, matched = compute_keyword_score("bitcoin etf fed gold oil", keywords) + assert score == 1.0 + assert len(matched) == 5 + def test_case_insensitive(self): score, matched = compute_keyword_score("BITCOIN ETF", ["bitcoin", "etf"]) - assert score == 1.0 + assert score == pytest.approx(2 / 3, abs=0.001) + assert set(matched) == {"bitcoin", "etf"} def test_score_rounded_to_4dp(self): score, _ = compute_keyword_score("bitcoin", ["bitcoin", "etf", "fed"]) diff --git a/skills/financial_news/tests/test_fetch_news.py b/skills/financial_news/tests/test_fetch_news.py index cee85c3..bc76613 100644 --- a/skills/financial_news/tests/test_fetch_news.py +++ b/skills/financial_news/tests/test_fetch_news.py @@ -13,7 +13,7 @@ sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) import fetch_news -from financial_news.fetcher import fetch_feed, strip_html +from financial_news.fetcher import dedup_articles, fetch_feed, strip_html @pytest.fixture(autouse=True) @@ -31,16 +31,27 @@ def reset_logger(): def _make_entry(title: str = "Test", published: time.struct_time = FUTURE, no_date: bool = False): pub = None if no_date else published + slug = title.lower().replace(" ", "-") entry = MagicMock() entry.get = lambda key, default="": { "title": title, "summary": f"{title} summary", - "link": "https://example.com/article", + "link": f"https://example.com/{slug}", "published_parsed": pub, }.get(key, default) return entry +def _article(title: str = "Test", url: str | None = None, published_at: str = "2026-01-01") -> dict: + return { + "title": title, + "summary": f"{title} summary", + "url": url if url is not None else f"https://example.com/{title.lower()}", + "published_at": published_at, + "source": "Feed", + } + + def _make_feed(entries: list | None = None, bozo: bool = False): feed = MagicMock() feed.bozo = bozo @@ -108,7 +119,7 @@ def test_fresh_entry_included_with_correct_fields(self): r = result[0] assert r["title"] == "Fresh" assert r["source"] == "TestFeed" - assert r["url"] == "https://example.com/article" + assert r["url"] == "https://example.com/fresh" assert "Fresh summary" in r["summary"] assert "published_at" in r @@ -132,6 +143,44 @@ def test_multiple_fresh_entries_all_returned(self): assert len(result) == 3 +# ── dedup_articles ──────────────────────────────────────────────────────────── + + +class TestDedupArticles: + def test_empty_list(self): + assert dedup_articles([]) == [] + + def test_unique_articles_untouched(self): + articles = [_article("One"), _article("Two"), _article("Three")] + assert dedup_articles(articles) == articles + + def test_same_url_dropped(self): + first = _article("Fed cuts rates", url="https://example.com/a") + copy = _article("Fed cuts rates (updated)", url="https://example.com/a") + assert dedup_articles([first, copy]) == [first] + + def test_url_match_ignores_case_and_trailing_slash(self): + first = _article("One", url="https://Example.com/Story/") + copy = _article("Two", url="https://example.com/story") + assert dedup_articles([first, copy]) == [first] + + def test_same_normalised_title_dropped(self): + first = _article("Fed Cuts Rates!", url="https://a.com/1") + copy = _article("fed cuts rates", url="https://b.com/2") + assert dedup_articles([first, copy]) == [first] + + def test_first_occurrence_kept(self): + first = _article("Story", url="https://a.com/1") + second = _article("Story", url="https://b.com/2") + result = dedup_articles([first, second]) + assert result == [first] + + def test_missing_url_and_title_not_treated_as_duplicates(self): + a = {"title": "", "url": "", "summary": "x", "published_at": "", "source": "F"} + b = {"title": "", "url": "", "summary": "y", "published_at": "", "source": "F"} + assert dedup_articles([a, b]) == [a, b] + + # ── main ────────────────────────────────────────────────────────────────────── _SINGLE_FEED = {"TestFeed": "http://x.com/feed"} diff --git a/skills/trader/src/trader/client.py b/skills/trader/src/trader/client.py index db8d6a0..8780dda 100644 --- a/skills/trader/src/trader/client.py +++ b/skills/trader/src/trader/client.py @@ -36,7 +36,12 @@ def get_client() -> Anthropic: def call_role( - client: Anthropic, model: str, system: str, user_message: str, max_tokens: int + client: Anthropic, + model: str, + system: str, + user_message: str, + max_tokens: int, + output_schema: dict | None = None, ) -> tuple[str, dict]: """ Send one stateless message to a role and return its text plus token usage. @@ -51,6 +56,8 @@ def call_role( system: The role's system prompt. user_message: The full context for this call. max_tokens: Response token cap for this role. + output_schema: Optional JSON schema. When given, the API constrains the + response to valid JSON matching it (structured outputs). Returns: ``(text, usage)`` — the response text and a per-model usage dict @@ -61,12 +68,15 @@ def call_role( f"call_role: model={model!r}, max_tokens={max_tokens}, prompt_chars={len(user_message)}" ) - response = client.messages.create( - model=model, - max_tokens=max_tokens, - system=system, - messages=[{"role": "user", "content": user_message}], - ) + request: dict = { + "model": model, + "max_tokens": max_tokens, + "system": system, + "messages": [{"role": "user", "content": user_message}], + } + if output_schema is not None: + request["output_config"] = {"format": {"type": "json_schema", "schema": output_schema}} + response = client.messages.create(**request) text = "".join(block.text for block in response.content if block.type == "text") used = record(model, getattr(response, "usage", None)) diff --git a/skills/trader/src/trader/judge.py b/skills/trader/src/trader/judge.py index d87dcd0..e51793f 100644 --- a/skills/trader/src/trader/judge.py +++ b/skills/trader/src/trader/judge.py @@ -8,6 +8,61 @@ from trader.client import call_role +_SIDE_SCORE_SCHEMA = { + "type": "object", + "properties": { + "evidence_quality": {"type": "integer"}, + "rebuttal_validity": {"type": "integer"}, + "risk_adjusted_logic": {"type": "integer"}, + "internal_consistency": {"type": "integer"}, + "falsifiability": {"type": "integer"}, + "total": {"type": "integer"}, + }, + "required": [ + "evidence_quality", + "rebuttal_validity", + "risk_adjusted_logic", + "internal_consistency", + "falsifiability", + "total", + ], + "additionalProperties": False, +} + +# Enforced via structured outputs so the judge cannot return malformed JSON. +# Mirrors the response shape documented in config/roles.yaml. +VERDICT_SCHEMA = { + "type": "object", + "properties": { + "scorecard": { + "type": "object", + "properties": {"bull": _SIDE_SCORE_SCHEMA, "bear": _SIDE_SCORE_SCHEMA}, + "required": ["bull", "bear"], + "additionalProperties": False, + }, + "winner": {"type": "string", "enum": ["bull", "bear", "split"]}, + "confidence": {"type": "string", "enum": ["high", "medium", "low"]}, + "strongest_bull_argument": {"type": "string"}, + "strongest_bear_argument": {"type": "string"}, + "bull_fatal_flaw": {"type": "string"}, + "bear_fatal_flaw": {"type": "string"}, + "key_unresolved_question": {"type": "string"}, + "verdict": {"type": "string"}, + }, + "required": [ + "scorecard", + "winner", + "confidence", + "strongest_bull_argument", + "strongest_bear_argument", + "bull_fatal_flaw", + "bear_fatal_flaw", + "key_unresolved_question", + "verdict", + ], + "additionalProperties": False, +} + def format_transcript(transcript: list[dict]) -> str: """Render the debate transcript as text for the judge prompt.""" @@ -56,5 +111,12 @@ def run_judge( logger.debug(f"run_judge: scoring {len(transcript)} transcript entries on {model!r}") prompt = f"Stock: {symbol}\n\nDEBATE TRANSCRIPT:\n{format_transcript(transcript)}" - text, used = call_role(client, model, config["judge"], prompt, config["max_tokens"]["judge"]) + text, used = call_role( + client, + model, + config["judge"], + prompt, + config["max_tokens"]["judge"], + output_schema=VERDICT_SCHEMA, + ) return parse_verdict(text), used diff --git a/skills/trader/tests/test_run_judge.py b/skills/trader/tests/test_run_judge.py index 0d6e992..34914c3 100644 --- a/skills/trader/tests/test_run_judge.py +++ b/skills/trader/tests/test_run_judge.py @@ -9,7 +9,7 @@ from unittest.mock import MagicMock, patch import pytest -from trader.judge import format_transcript, parse_verdict, run_judge +from trader.judge import VERDICT_SCHEMA, format_transcript, parse_verdict, run_judge _spec = importlib.util.spec_from_file_location( "trader_run_judge", Path(__file__).parent.parent / "scripts" / "run_judge.py" @@ -113,6 +113,26 @@ def test_uses_judge_model_and_system(self): assert call.kwargs["model"] == "claude-opus-4-8" assert call.kwargs["system"] == "You are the judge." + def test_enforces_verdict_schema_via_structured_outputs(self): + client = MagicMock() + client.messages.create.return_value = _fake_response("{}") + + run_judge(client, SAMPLE_CONFIG, "AMZN", SAMPLE_TRANSCRIPT) + + output_config = client.messages.create.call_args.kwargs["output_config"] + assert output_config == {"format": {"type": "json_schema", "schema": VERDICT_SCHEMA}} + + def test_verdict_schema_objects_forbid_extra_properties(self): + # Structured outputs requires additionalProperties: false on every object. + def _check(schema: dict): + if schema.get("type") == "object": + assert schema["additionalProperties"] is False + assert "required" in schema + for sub in schema["properties"].values(): + _check(sub) + + _check(VERDICT_SCHEMA) + def test_prompt_contains_symbol_and_transcript(self): client = MagicMock() client.messages.create.return_value = _fake_response("{}")