diff --git a/scalable_sync.py b/scalable_sync.py index 45b8dbc..0906674 100644 --- a/scalable_sync.py +++ b/scalable_sync.py @@ -34,6 +34,7 @@ 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") @@ -63,27 +64,65 @@ def sc(*args: str, timeout: int = 60) -> dict: 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.isoformat()] - if to_time: args += ["--to-time", to_time.isoformat()] + 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] + 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] + 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_transaction_details(tx_id: str) -> dict: - return sc("broker", "transaction", "details", "--transaction-id", tx_id) - - def sc_holdings() -> dict: - return sc("broker", "holdings") + raw = sc("broker", "holdings") + return raw.get("result", raw) # --------------------------------------------------------------------------- @@ -173,18 +212,17 @@ def audit(action, payload): def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]: - tx_type = tx.get("type") # "SECURITY_TRANSACTION" | "CASH_TRANSACTION" + tx_type = tx.get("type") status = (tx.get("status") or "").upper() - - if tx.get("is_cancellation"): - return [] - if SC_SETTLED_ONLY and status != "SETTLED": - return [] - 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": @@ -194,10 +232,8 @@ def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]: dead_letter(tx, f"unmapped ISIN: {isin}") return [] - side = (tx.get("side") or "").upper() # "BUY" | "SELL" + side = (tx.get("side") or "").upper() qty = tx.get("quantity") - - # unit_price aus amount/quantity ableiten unit_price = round(amount / qty, 6) if qty else None activities.append(WealthfolioActivity( @@ -207,8 +243,6 @@ def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]: quantity=qty, unit_price=unit_price, amount=amount, - fee=None, - tax=None, external_id=tx_id, account_id=WF_ACCOUNT_ID, )) @@ -222,8 +256,8 @@ def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]: "FEE": "FEE", "TAX": "TAX", "TAX_RETURN": "CREDIT", - "CASH_TRANSFER_IN": "DEPOSIT", # neu - "CASH_TRANSFER_OUT": "WITHDRAWAL", # neu + "CASH_TRANSFER_IN": "DEPOSIT", + "CASH_TRANSFER_OUT": "WITHDRAWAL", }.get(cash_type) if not wf_type: @@ -234,7 +268,7 @@ def map_to_wealthfolio(tx: dict) -> list[WealthfolioActivity]: date=last_event, symbol="$CASH", activity_type=wf_type, - amount=abs(tx.get("amount") or 0), + amount=amount, external_id=tx_id, account_id=WF_ACCOUNT_ID, )) @@ -312,13 +346,15 @@ class WealthfolioMCP: } for activity in activities ] - result = asyncio.run( self._connect_and_call("record_activities", {"activities": payload}) ) - log.info("MCP record_activities result: %s", result) # ← neu + if getattr(result, "is_error", False): + raise RuntimeError(f"record_activities error: {result.content}") + log.info("record_activities OK: %s activities", len(activities)) return result + # --------------------------------------------------------------------------- # Sync # --------------------------------------------------------------------------- @@ -336,11 +372,11 @@ def reconcile(mcp): def sync_once(dry_run=False): init_db() - mcp = None + 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") @@ -351,8 +387,8 @@ def sync_once(dry_run=False): 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") + items = page["items"] + cursor = page["cursor"] for summary in items: tx_id = summary.get("id") @@ -365,9 +401,6 @@ def sync_once(dry_run=False): continue activities = map_to_wealthfolio(summary) - if activities: - all_activities.extend(activities) - mark_imported(...) if not activities: stats["dead_letter"] += 1 continue @@ -377,18 +410,17 @@ def sync_once(dry_run=False): 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) # ← diese Zeile fehlt noch + stats["new"] += len(activities) else: try: - result = mcp.record_activities(activities) + 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 all_activities and not dry_run: - mcp.record_activities(all_activities) + if not cursor or not items: break