#!/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 time from dataclasses import dataclass, asdict from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Optional import asyncio import httpx from mcp import ClientSession from mcp.client.streamable_http import streamable_http_client # --------------------------------------------------------------------------- # 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", "{}")) SC_PORTFOLIO_ID = os.getenv("SC_PORTFOLIO_ID") 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/appuser", "XDG_CONFIG_HOME": "/home/appuser/.config"} 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_ensure_context(): """ Check if a broker context is set. If not (or if SC_PORTFOLIO_ID is configured), call 'sc broker context select' to set it. Falls back to 'sc login' if needed. """ # If a portfolio ID is explicitly configured, always select it if SC_PORTFOLIO_ID: log.info("Setting broker context to portfolio %s", SC_PORTFOLIO_ID) try: sc("broker", "context", "select", "--portfolio-id", SC_PORTFOLIO_ID) log.info("Broker context set via SC_PORTFOLIO_ID") return except Exception as e: log.warning("Failed to set context via SC_PORTFOLIO_ID: %s", e) # Check if context already exists try: ctx_raw = sc("broker", "context", "show") ctx = ctx_raw.get("result", ctx_raw) if isinstance(ctx_raw, dict) else {} portfolio_id = ctx.get("portfolio_id") or ctx_raw.get("portfolio_id") if portfolio_id: log.info("Broker context already set: portfolio=%s", portfolio_id) return except Exception: pass # No context — need to login log.info("No broker context found, attempting sc login...") env = {**os.environ, "HOME": "/home/appuser", "XDG_CONFIG_HOME": "/home/appuser/.config"} result = subprocess.run(["sc", "login"], env=env, timeout=120) if result.returncode != 0: raise RuntimeError("sc login failed — cannot proceed without broker context") log.info("sc login completed") 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.strftime("%Y-%m-%dT%H:%M:%SZ")] if to_time: args += ["--to-time", to_time.strftime("%Y-%m-%dT%H:%M:%SZ")] 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] raw = sc(*args) result = raw.get("result", raw) return {"items": result.get("items", []), "cursor": result.get("cursor")} def sc_holdings() -> dict: raw = sc("broker", "holdings") return raw.get("result", raw) # --------------------------------------------------------------------------- # 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", "side", "security_transaction_type", "cash_transaction_type", "amount", "quantity", "isin", ]} 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 activity_type: str currency: str = "EUR" symbol: Optional[str] = None quantity: Optional[float] = None unit_price: Optional[float] = None amount: Optional[float] = None fee: Optional[float] = None tax: Optional[float] = None 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() last_event = (tx.get("last_event_datetime") or "")[:10] tx_id = tx.get("id", "") amount = abs(tx.get("amount") or 0) if tx.get("is_cancellation"): return [] if SC_SETTLED_ONLY and status not in {"SETTLED", "CONFIRMED"}: return [] activities = [] if tx_type == "SECURITY_TRANSACTION": isin = tx.get("isin") symbol = ASSET_MAP.get(isin) if isin else None if not symbol: dead_letter(tx, f"unmapped ISIN: {isin}") return [] side = (tx.get("side") or "").upper() qty = tx.get("quantity") unit_price = round(amount / qty, 6) if qty else None activities.append(WealthfolioActivity( date=last_event, symbol=symbol, activity_type="BUY" if side == "BUY" else "SELL", quantity=qty, unit_price=unit_price, amount=amount, currency=tx.get("currency", "EUR"), external_id=tx_id, account_id=WF_ACCOUNT_ID, )) elif tx_type == "CASH_TRANSACTION": cash_type = (tx.get("cash_transaction_type") or "").upper() wf_type = { "DEPOSIT": "DEPOSIT", "WITHDRAWAL": "WITHDRAWAL", "INTEREST": "INTEREST", "FEE": "FEE", "TAX": "TAX", "TAX_RETURN": "CREDIT", "CASH_TRANSFER_IN": "DEPOSIT", "CASH_TRANSFER_OUT": "WITHDRAWAL", }.get(cash_type) if not wf_type: dead_letter(tx, f"unknown cash_transaction_type: {cash_type}") return [] activities.append(WealthfolioActivity( date=last_event, symbol=None, activity_type=wf_type, amount=amount, currency=tx.get("currency", "EUR"), external_id=tx_id, account_id=WF_ACCOUNT_ID, )) else: dead_letter(tx, f"unknown type: {tx_type}") return [] return activities # --------------------------------------------------------------------------- # Wealthfolio MCP Client # --------------------------------------------------------------------------- class WealthfolioMCP: def __init__(self, url: str, token: str): self.url = url self.token = token def _http_client(self) -> httpx.AsyncClient: return httpx.AsyncClient( headers={ "Authorization": f"Bearer {self.token}", "Accept": "application/json, text/event-stream", }, timeout=30.0, ) async def _connect_and_call(self, tool_name: str, arguments: dict): async with self._http_client() as http_client: async with streamable_http_client( self.url, http_client=http_client, ) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: await session.initialize() return await session.call_tool(tool_name, arguments) async def _list_tools(self) -> list[str]: async with self._http_client() as http_client: async with streamable_http_client( self.url, http_client=http_client, ) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: await session.initialize() result = await session.list_tools() return [tool.name for tool in result.tools] def ping(self) -> bool: try: tools = asyncio.run(self._list_tools()) log.info("Available Wealthfolio MCP tools: %s", tools) return True except Exception as exc: log.exception("MCP initialization failed: %s", exc) return False def record_activities(self, activities): activity_payloads = [] for activity in activities: payload = { "account": activity.account_id, "activityDate": activity.date, "activityType": activity.activity_type, "amount": activity.amount, "currency": activity.currency, "fee": activity.fee, "tax": activity.tax, "notes": f"scalable:{activity.external_id}", } # Nur asset-basierte Aktivitäten brauchen Symbol, Menge und Preis. if activity.activity_type in {"BUY", "SELL", "DIVIDEND", "SPLIT"}: payload["symbol"] = activity.symbol payload["quantity"] = activity.quantity payload["unitPrice"] = activity.unit_price activity_payloads.append( {key: value for key, value in payload.items() if value is not None} ) draft_result = asyncio.run( self._connect_and_call( "record_activities", {"activities": activity_payloads}, ) ) if getattr(draft_result, "is_error", False): raise RuntimeError( f"record_activities error: {draft_result.content}" ) draft_data = draft_result.structured_content or {} draft_rows = draft_data.get("drafts", []) if len(draft_rows) != len(activities): raise RuntimeError( "Wealthfolio returned an incomplete draft batch: " f"expected {len(activities)}, got {len(draft_rows)}; " + json.dumps(draft_data, default=str) ) invalid_rows = [ row for row in draft_rows if row.get("errors") or not row.get("validation", {}).get("isValid", False) or not row.get("draft", {}).get("accountId") ] if invalid_rows: raise RuntimeError( "Wealthfolio returned invalid activity drafts: " + json.dumps(invalid_rows, default=str) ) drafts = [row["draft"] for row in draft_rows] commit_result = asyncio.run( self._connect_and_call( "commit_activity_drafts", {"drafts": drafts}, ) ) if getattr(commit_result, "is_error", False): raise RuntimeError( f"commit_activity_drafts error: {commit_result.content}" ) commit_data = commit_result.structured_content or {} if commit_data.get("errors"): raise RuntimeError( "Wealthfolio failed to persist activities: " + json.dumps(commit_data["errors"], default=str) ) created = commit_data.get("created", []) if len(created) != len(activities): raise RuntimeError( "Wealthfolio commit was incomplete: " f"expected {len(activities)}, got {len(created)}; " + json.dumps(commit_data, default=str) ) log.info("Wealthfolio committed %s activities", len(created)) return created # --------------------------------------------------------------------------- # 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() sc_ensure_context() mcp = None if not dry_run: 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["items"] cursor = page["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 activities = map_to_wealthfolio(summary) 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]}) stats["new"] += len(activities) else: try: mcp.record_activities(activities) mark_imported(tx_id, tx_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 if not dry_run: 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()