525 lines
18 KiB
Python
525 lines
18 KiB
Python
#!/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
|
|
symbol: Optional[str] = None
|
|
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()
|
|
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() |