diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..9f3e99f --- /dev/null +++ b/Dockerfile @@ -0,0 +1,41 @@ +# syntax=docker/dockerfile:1 +FROM python:3.13-slim-bookworm AS builder + +ARG SCALABLE_CLI_VERSION=0.1.0 +ARG TARGETARCH + +RUN apt-get update && apt-get install -y --no-install-recommends curl ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +RUN ARCH=$([ "$TARGETARCH" = "arm64" ] && echo "aarch64" || echo "x86_64") \ + && curl -fsSL -o /tmp/sc.tar.gz \ + "https://github.com/ScalableCapital/scalable-cli/releases/download/v${SCALABLE_CLI_VERSION}/scalable-cli-${ARCH}-unknown-linux-gnu.tar.gz" \ + && curl -fsSL -o /tmp/sc.tar.gz.sha256 \ + "https://github.com/ScalableCapital/scalable-cli/releases/download/v${SCALABLE_CLI_VERSION}/scalable-cli-${ARCH}-unknown-linux-gnu.tar.gz.sha256" \ + && cd /tmp && sha256sum -c sc.tar.gz.sha256 \ + && tar -xzf sc.tar.gz \ + && mv scalable-cli-${ARCH}-unknown-linux-gnu/sc /usr/local/bin/sc \ + && chmod +x /usr/local/bin/sc + +FROM python:3.13-slim-bookworm + +ARG PUID=1000 +ARG PGID=1000 + +# User mit konfigurierbarer UID/GID anlegen +RUN groupadd -g ${PGID} appgroup \ + && useradd -u ${PUID} -g appgroup -m -d /home/appuser appuser + +COPY --from=builder /usr/local/bin/sc /usr/local/bin/sc + +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY scalable_sync.py . + +ENV SC_CONFIG_DIR=/tmp/sc-config + +USER ${PUID}:${PGID} + +ENTRYPOINT ["python", "/app/scalable_sync.py"] \ No newline at end of file diff --git a/README.md b/README.md index 556ac13..d3ca3c5 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,111 @@ -# scalable-wealth +# scalable-sync +Synchronisiert Scalable Capital Broker-Transaktionen nach Wealthfolio. + +## Voraussetzungen + +- Docker + Docker Compose +- Wealthfolio mit MCP aktiviert (`WF_MCP_ENABLED=true`) +- Scalable Capital: Agentic Investing aktiviert (Profil → Sicherheit) + +## Dateien + +| Datei | Zweck | +|---|---| +| `Dockerfile` | Container-Image | +| `compose.yaml` | Docker Compose Config | +| `.env` | Alle Einstellungen | +| `scalable_sync.py` | Sync-Adapter | +| `requirements.txt` | Python-Abhängigkeiten | + +## Setup + +```bash +# Dateien in ein Verzeichnis deiner Wahl kopieren, z. B.: +# /opt/scalable-sync/ oder ~/docker/scalable-sync/ + +docker compose build + +# 1. Einmalig einloggen (Session wird im Volume gespeichert) +docker compose run --rm -it scalable-sync sc login --local-read-only + +# 2. Test +docker compose run --rm scalable-sync sc whoami +docker compose run --rm scalable-sync sc broker holdings --json | jq . + +# 3. Dry-Run +docker compose run --rm scalable-sync python /app/scalable_sync.py --once --dry-run + +# 4. Start +docker compose up -d +``` + +## .env + +```bash +# Wealthfolio +WF_MCP_URL=http://wealthfolio:8088/mcp +WF_MCP_TOKEN=wf_mcp_dein_token_hier +WF_ACCOUNT_ID=dein-konto-uuid + +# Optional: Scalable Portfolio-ID (bei mehreren Portfolios) +SC_PORTFOLIO_ID= + +# Sync +SYNC_INTERVAL_HOURS=6 +SC_OVERLAP_DAYS=14 +SC_SETTLED_ONLY=true + +# ISIN → Wealthfolio-Symbol +ASSET_MAP={"IE00B4L5Y983":"EUNL.DE","IE00B3RBWM25":"VWCE.DE"} + +# Container-User +PUID=1000 +PGID=1000 +``` + +### WF_ACCOUNT_ID finden + +Wealthfolio → Accounts → Konto anklicken → ID aus der URL kopieren (`/accounts/abc123-def456/...`). + +### ASSET_MAP + +JSON-Objekt: `"ISIN": "SYMBOL"`. Wealthfolio-Symbol unter Assets suchen. + +### SC_PORTFOLIO_ID + +Nur bei mehreren Scalable-Portfolios nötig. Ermitteln mit: +```bash +docker compose run --rm scalable-sync sc broker context show --json +``` + +## Volumes + +Docker legt diese Volumes automatisch an — kein manuelles `mkdir` nötig. + +| Volume | Inhalt | Zweck | +|---|---|---| +| `scalable-sync-data` | state.db, audit.jsonl, dead-letter.jsonl | Sync-Status und Logs | +| `scalable-sync-config` | config.toml, session.json | Scalable CLI Session | + +**Wichtig:** Ohne `scalable-sync-config` geht die Session bei `docker compose down` verloren. + +## Befehle + +| Befehl | Beschreibung | +|---|---| +| `docker compose run --rm scalable-sync python /app/scalable_sync.py --once --dry-run` | Testlauf ohne Schreiben | +| `docker compose run --rm scalable-sync python /app/scalable_sync.py --once` | Einmaliger Sync | +| `docker compose logs -f` | Logs anzeigen | +| `docker compose down && docker compose up -d` | Neustart (Session bleibt erhalten) | + +## Troubleshooting + +| Problem | Lösung | +|---|---| +| "Not logged in" | `docker compose run --rm -it scalable-sync sc login --local-read-only` | +| Keine Transaktionen | `SC_SETTLED_ONLY` und `SC_OVERLAP_DAYS` prüfen | +| Unmapped ISIN | ISIN in `ASSET_MAP` ergänzen | +| Mehrere Portfolios | `SC_PORTFOLIO_ID` setzen | +| Doppelte Activities | Prüfen ob `WF_ACCOUNT_ID` korrekt ist | +| Session nach Neustart weg | `scalable-sync-config` Volume in compose.yaml prüfen | \ No newline at end of file diff --git a/compose.yaml b/compose.yaml new file mode 100644 index 0000000..9f95f46 --- /dev/null +++ b/compose.yaml @@ -0,0 +1,34 @@ +services: + scalable-sync: + build: + context: . + args: + SCALABLE_CLI_VERSION: "0.1.0" + PUID: ${PUID} + PGID: ${PGID} + image: scalable-sync:latest + hostname: scalable-sync + container_name: scalable-sync + restart: unless-stopped + user: "${PUID}:${PGID}" + networks: + - ext_net + volumes: + - scalable-sync-data:/app/data + - scalable-sync-config:/tmp/sc-config + env_file: + - .env + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request,json,os; req=urllib.request.Request(os.environ['WF_MCP_URL'],method='POST',data=json.dumps({'jsonrpc':'2.0','id':1,'method':'ping'}).encode(),headers={'Authorization':'Bearer '+os.environ['WF_MCP_TOKEN'],'Content-Type':'application/json'}); urllib.request.urlopen(req,timeout=5)"] + interval: 5m + timeout: 10s + retries: 3 + start_period: 30s + +networks: + ext_net: + external: true + +volumes: + scalable-sync-data: + scalable-sync-config: \ No newline at end of file diff --git a/example.env b/example.env new file mode 100644 index 0000000..d5145e5 --- /dev/null +++ b/example.env @@ -0,0 +1,16 @@ +# ── Wealthfolio ────────────────────────────────────────────── +WF_MCP_URL=http://wealthfolio:8088/mcp +WF_MCP_TOKEN=wf_mcp_dein_token_hier + +# ── Sync-Intervalle ────────────────────────────────────────── +SYNC_INTERVAL_HOURS=6 +SC_OVERLAP_DAYS=14 +SC_SETTLED_ONLY=true +SC_MAPPING_VERSION=1 + +# ── Asset-Mapping ────────── +ASSET_MAP={"IE00B4L5Y983":"EUNL.DE","IE00B3RBWM25":"VWCE.DE","DE0002635307":"EXSA.DE"} + +# ── User im Container ───────── +PUID=1000 +PGID=1000 \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..6f89372 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +requests>=2.32.0 \ No newline at end of file diff --git a/scalable_sync.py b/scalable_sync.py new file mode 100644 index 0000000..9c50327 --- /dev/null +++ b/scalable_sync.py @@ -0,0 +1,398 @@ +#!/usr/bin/env python3 +""" +scalable-sync: Scalable Capital → Wealthfolio MCP Sync Adapter +Alle Config über Environment-Variablen. +""" + +import json +import logging +import os +import sqlite3 +import subprocess +import sys +import tempfile +import time +from dataclasses import dataclass, asdict +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Optional + +import requests + +# --------------------------------------------------------------------------- +# Config +# --------------------------------------------------------------------------- + +WF_MCP_URL = os.environ["WF_MCP_URL"] +WF_MCP_TOKEN = os.environ["WF_MCP_TOKEN"] +WF_ACCOUNT_ID = os.environ.get("WF_ACCOUNT_ID") +SYNC_INTERVAL_HOURS = int(os.getenv("SYNC_INTERVAL_HOURS", "6")) +SC_OVERLAP_DAYS = int(os.getenv("SC_OVERLAP_DAYS", "14")) +SC_SETTLED_ONLY = os.getenv("SC_SETTLED_ONLY", "true").lower() == "true" +SC_MAPPING_VERSION = int(os.getenv("SC_MAPPING_VERSION", "1")) +ASSET_MAP = json.loads(os.getenv("ASSET_MAP", "{}")) + +# Scalable CLI Config: tmpfs-basiert, automatisch generiert +SC_CONFIG_DIR = Path(os.getenv("SC_CONFIG_DIR", tempfile.mkdtemp(prefix="sc-config-"))) +SC_CONFIG_DIR.mkdir(parents=True, exist_ok=True) +(SC_CONFIG_DIR / "config.toml").write_text( + '[auth]\nsession_backend = "file"\nsigning_key_backend = "file"\n' +) +os.environ["SC_CONFIG_DIR"] = str(SC_CONFIG_DIR) + +DATA_DIR = Path("/app/data") +DATA_DIR.mkdir(exist_ok=True) +DB_PATH = DATA_DIR / "state.db" +AUDIT_PATH = DATA_DIR / "audit.jsonl" +DEAD_LETTER_PATH = DATA_DIR / "dead-letter.jsonl" + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + handlers=[logging.StreamHandler(sys.stdout)] +) +log = logging.getLogger("scalable-sync") + +# --------------------------------------------------------------------------- +# Scalable CLI Wrapper +# --------------------------------------------------------------------------- + +def sc(*args: str, timeout: int = 60) -> dict: + cmd = ["sc"] + list(args) + env = {**os.environ, "HOME": "/home/sync", "SC_CONFIG_DIR": str(SC_CONFIG_DIR)} + result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, env=env) + if result.returncode != 0: + log.error("sc failed: %s", result.stderr) + raise RuntimeError(f"sc exit {result.returncode}: {result.stderr}") + return json.loads(result.stdout) + + +def sc_transactions(from_time=None, to_time=None, type_filter=None, status=None, isin=None, cursor=None, page_size=100): + args = ["broker", "transactions", "--page-size", str(page_size)] + if from_time: args += ["--from-time", from_time.isoformat()] + if to_time: args += ["--to-time", to_time.isoformat()] + if type_filter: + for t in type_filter: args += ["--type-filter", t] + if status: + for s in status: args += ["--status", s] + if isin: args += ["--isin", isin] + if cursor: args += ["--cursor", cursor] + return sc(*args) + + +def sc_transaction_details(tx_id: str) -> dict: + return sc("broker", "transaction", "details", "--transaction-id", tx_id) + + +def sc_holdings() -> dict: + return sc("broker", "holdings") + + +# --------------------------------------------------------------------------- +# State +# --------------------------------------------------------------------------- + +def init_db(): + with sqlite3.connect(DB_PATH) as conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS sync_state ( + scalable_transaction_id TEXT PRIMARY KEY, + source_hash TEXT NOT NULL, + source_status TEXT NOT NULL, + wealthfolio_activity_ids TEXT, + imported_at TEXT, + last_seen_at TEXT NOT NULL, + mapping_version INTEGER NOT NULL + ) + """) + conn.execute("CREATE INDEX IF NOT EXISTS idx_last_seen ON sync_state(last_seen_at)") + + +def hash_transaction(tx: dict) -> str: + relevant = {k: tx.get(k) for k in [ + "id", "type", "status", "last_event_datetime", "security", + "security_trade", "cash", "non_trade_security" + ]} + return json.dumps(relevant, sort_keys=True, default=str) + + +def is_imported(tx_id: str, current_hash: str) -> bool: + with sqlite3.connect(DB_PATH) as conn: + row = conn.execute("SELECT source_hash FROM sync_state WHERE scalable_transaction_id = ?", (tx_id,)).fetchone() + return row is not None and row[0] == current_hash + + +def mark_imported(tx_id, tx_hash, status, wf_ids): + now = datetime.now(timezone.utc).isoformat() + with sqlite3.connect(DB_PATH) as conn: + conn.execute(""" + INSERT OR REPLACE INTO sync_state VALUES (?, ?, ?, ?, ?, ?, ?) + """, (tx_id, tx_hash, status, json.dumps(wf_ids), now, now, SC_MAPPING_VERSION)) + + +def mark_seen(tx_id, tx_hash, status): + now = datetime.now(timezone.utc).isoformat() + with sqlite3.connect(DB_PATH) as conn: + conn.execute(""" + INSERT OR IGNORE INTO sync_state (scalable_transaction_id, source_hash, source_status, last_seen_at, mapping_version) + VALUES (?, ?, ?, ?, ?) + """, (tx_id, tx_hash, status, now, SC_MAPPING_VERSION)) + conn.execute("UPDATE sync_state SET last_seen_at = ? WHERE scalable_transaction_id = ?", (now, tx_id)) + + +# --------------------------------------------------------------------------- +# Mapping +# --------------------------------------------------------------------------- + +@dataclass +class WealthfolioActivity: + date: str + symbol: str + activity_type: str + quantity: Optional[float] = None + unit_price: Optional[float] = None + amount: Optional[float] = None + fee: Optional[float] = None + tax: Optional[float] = None + currency: str = "EUR" + source: str = "scalable-cli" + external_id: Optional[str] = None + account_id: Optional[str] = None + + +def dead_letter(tx, reason): + entry = {"timestamp": datetime.now(timezone.utc).isoformat(), "reason": reason, "transaction": tx} + with open(DEAD_LETTER_PATH, "a") as f: + f.write(json.dumps(entry, default=str) + "\n") + log.warning("Dead letter: %s (%s)", tx.get("id"), reason) + + +def audit(action, payload): + entry = {"timestamp": datetime.now(timezone.utc).isoformat(), "action": action, "payload": payload} + with open(AUDIT_PATH, "a") as f: + f.write(json.dumps(entry, default=str) + "\n") + + +def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]: + tx_type = tx.get("type") + status = (tx.get("status") or "").upper() + if tx.get("is_cancellation") or status in {"CANCELLED", "EXPIRED", "REJECTED", "CANCEL_REQUESTED"}: + return [] + if SC_SETTLED_ONLY and status not in {"SETTLED", "CONFIRMED"}: + return [] + if tx.get("is_pending"): + return [] + + sec = tx.get("security") or {} + isin = sec.get("isin") + symbol = ASSET_MAP.get(isin) if isin else None + + st = tx.get("security_trade") or {} + cash = tx.get("cash") or {} + nts = tx.get("non_trade_security") or {} + last_event = (tx.get("last_event_datetime") or "")[:10] + tx_id = tx.get("id", "") + + activities = [] + + if tx_type in {"BUY", "SELL", "SAVINGS_PLAN", "REINVESTMENT"}: + if not symbol: + dead_letter(tx, f"unmapped ISIN: {isin}") + return [] + qty = (st.get("number_of_shares") or {}).get("filled") or nts.get("quantity") + price = st.get("average_price") or nts.get("average_price") + amount = st.get("total_amount") or nts.get("total_amount") + fee = st.get("fee") or st.get("transactional_fee") + tax = (st.get("aggregated_transaction_taxes") or {}).get("total_tax") + activities.append(WealthfolioActivity( + date=last_event, symbol=symbol, + activity_type="BUY" if tx_type in {"BUY", "SAVINGS_PLAN", "REINVESTMENT"} else "SELL", + quantity=qty, unit_price=price, amount=amount, fee=fee, tax=tax, + external_id=tx_id, account_id=WF_ACCOUNT_ID + )) + + elif tx_type == "DEPOSIT": + activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="DEPOSIT", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) + elif tx_type == "WITHDRAWAL": + activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="WITHDRAWAL", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) + elif tx_type == "DISTRIBUTION": + if not symbol: + dead_letter(tx, "distribution without ISIN") + return [] + activities.append(WealthfolioActivity(date=last_event, symbol=symbol, activity_type="DIVIDEND", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) + elif tx_type == "INTEREST": + activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="INTEREST", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) + elif tx_type == "FEE": + activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="FEE", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) + elif tx_type == "TAX": + activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="TAX", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) + elif tx_type == "TAX_RETURN": + activities.append(WealthfolioActivity(date=last_event, symbol="$CASH", activity_type="CREDIT", amount=cash.get("amount"), external_id=tx_id, account_id=WF_ACCOUNT_ID)) + elif tx_type in {"TRANSFER_IN", "CASH_TRANSFER_IN"}: + target = "TRANSFER_IN" if symbol else "DEPOSIT" + activities.append(WealthfolioActivity( + date=last_event, symbol=symbol or "$CASH", activity_type=target, + quantity=nts.get("quantity"), amount=cash.get("amount") or nts.get("total_amount"), + external_id=tx_id, account_id=WF_ACCOUNT_ID + )) + elif tx_type in {"TRANSFER_OUT", "CASH_TRANSFER_OUT"}: + target = "TRANSFER_OUT" if symbol else "WITHDRAWAL" + activities.append(WealthfolioActivity( + date=last_event, symbol=symbol or "$CASH", activity_type=target, + quantity=nts.get("quantity"), amount=cash.get("amount") or nts.get("total_amount"), + external_id=tx_id, account_id=WF_ACCOUNT_ID + )) + elif tx_type in { + "SWAP_IN", "SWAP_OUT", "CURRENCY_SWITCH_BUY", "CURRENCY_SWITCH_SELL", + "POCKET_MONEY", "REINVESTMENT_DISTRIBUTION", "REINVESTMENT_POCKET_MONEY" + }: + dead_letter(tx, f"unsupported type: {tx_type}") + return [] + else: + dead_letter(tx, f"unknown type: {tx_type}") + return [] + + return activities + + +# --------------------------------------------------------------------------- +# Wealthfolio MCP Client +# --------------------------------------------------------------------------- + +class WealthfolioMCP: + def __init__(self, url, token): + self.url = url + self.session = requests.Session() + self.session.headers.update({ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + "Accept": "application/json", + }) + + def _rpc(self, method, params=None): + payload = {"jsonrpc": "2.0", "id": int(time.time() * 1000), "method": method} + if params: payload["params"] = params + resp = self.session.post(self.url, json=payload, timeout=30) + resp.raise_for_status() + return resp.json() + + def ping(self): + try: + return self._rpc("ping").get("result") == "pong" + except Exception as e: + log.error("MCP ping failed: %s", e) + return False + + def record_activities(self, activities): + payload = [{ + "accountId": a.account_id, "date": a.date, "symbol": a.symbol, "activityType": a.activity_type, + "quantity": a.quantity, "unitPrice": a.unit_price, "amount": a.amount, + "fee": a.fee, "tax": a.tax, "currency": a.currency, + "source": a.source, "externalId": a.external_id, + } for a in activities] + return self._rpc("tools/call", {"name": "record_activities", "arguments": {"activities": payload}}) + + +# --------------------------------------------------------------------------- +# Sync +# --------------------------------------------------------------------------- + +def reconcile(mcp): + try: + sc_h = sc_holdings() + log.info("Scalable holdings count: %s", len(sc_h.get("items", []))) + audit("reconcile", {"scalable_holdings": sc_h}) + return True + except Exception as e: + log.error("Reconcile failed: %s", e) + return False + + +def sync_once(dry_run=False): + init_db() + mcp = WealthfolioMCP(WF_MCP_URL, WF_MCP_TOKEN) + if not mcp.ping(): + raise RuntimeError("Wealthfolio MCP unreachable") + + from_time = datetime.now(timezone.utc) - timedelta(days=SC_OVERLAP_DAYS) + to_time = datetime.now(timezone.utc) + stats = {"new": 0, "skipped": 0, "dead_letter": 0, "error": 0} + cursor = None + + while True: + page = sc_transactions(from_time=from_time, to_time=to_time, cursor=cursor, page_size=100) + items = page.get("items", []) + cursor = page.get("cursor") + + for summary in items: + tx_id = summary.get("id") + tx_hash = hash_transaction(summary) + status = (summary.get("status") or "").upper() + + if is_imported(tx_id, tx_hash): + mark_seen(tx_id, tx_hash, status) + stats["skipped"] += 1 + continue + + try: + detail = sc_transaction_details(tx_id) + except Exception as e: + log.error("Failed details %s: %s", tx_id, e) + stats["error"] += 1 + continue + + detail_hash = hash_transaction(detail) + activities = map_to_wealthfolio(detail) + if not activities: + stats["dead_letter"] += 1 + continue + + if dry_run: + log.info("[DRY RUN] Would import %s: %s", tx_id, [ + f"{a.activity_type} {a.symbol} qty={a.quantity} amt={a.amount}" for a in activities + ]) + audit("dry_run", {"tx_id": tx_id, "activities": [asdict(a) for a in activities]}) + else: + try: + result = mcp.record_activities(activities) + mark_imported(tx_id, detail_hash, status, [str(i) for i in range(len(activities))]) + audit("imported", {"tx_id": tx_id, "activities": len(activities)}) + stats["new"] += len(activities) + except Exception as e: + log.error("Import failed %s: %s", tx_id, e) + stats["error"] += 1 + + if not cursor or not items: + break + + reconcile(mcp) + log.info("Sync complete: %s", stats) + return stats + + +def main(): + import argparse + p = argparse.ArgumentParser() + p.add_argument("--once", action="store_true") + p.add_argument("--dry-run", action="store_true") + p.add_argument("--daemon", action="store_true") + args = p.parse_args() + + if args.once or args.dry_run: + stats = sync_once(dry_run=args.dry_run) + sys.exit(0 if stats["error"] == 0 else 1) + + if args.daemon: + log.info("Starting daemon, interval=%dh", SYNC_INTERVAL_HOURS) + while True: + try: + sync_once() + except Exception as e: + log.exception("Sync failed: %s", e) + time.sleep(SYNC_INTERVAL_HOURS * 3600) + else: + p.print_help() + + +if __name__ == "__main__": + main() \ No newline at end of file