From a202be3fc14516de35f797399c1ce81e82ffec30 Mon Sep 17 00:00:00 2001 From: Tyler Wilcox Date: Sat, 23 May 2026 22:40:22 -0400 Subject: [PATCH] AI: Charles Schwab Account Adapter --- overview/brokerages/__init__.py | 2 + overview/brokerages/charles_schwab.py | 331 ++++++++++++++++++++++++++ overview/brokerages/registry.py | 2 + overview/tests.py | 102 ++++++++ 4 files changed, 437 insertions(+) create mode 100644 overview/brokerages/charles_schwab.py diff --git a/overview/brokerages/__init__.py b/overview/brokerages/__init__.py index 4d134e0..d77a0e8 100644 --- a/overview/brokerages/__init__.py +++ b/overview/brokerages/__init__.py @@ -1,10 +1,12 @@ from .base import BaseBrokerageAdapter, BrokerageAdapterError +from .charles_schwab import CharlesSchwabBrokerageAdapter from .registry import get_brokerage_adapter, register_brokerage_adapter from .random import RandomBrokerageAdapter __all__ = [ "BaseBrokerageAdapter", "BrokerageAdapterError", + "CharlesSchwabBrokerageAdapter", "RandomBrokerageAdapter", "get_brokerage_adapter", "register_brokerage_adapter", diff --git a/overview/brokerages/charles_schwab.py b/overview/brokerages/charles_schwab.py new file mode 100644 index 0000000..b356e08 --- /dev/null +++ b/overview/brokerages/charles_schwab.py @@ -0,0 +1,331 @@ +import json +from datetime import datetime, timedelta, timezone +from urllib.parse import urlencode, urljoin +from urllib.request import Request, urlopen + +from overview.brokerage_types import ( + BrokerageAccountPayload, + BrokerageAssetPayload, + BrokerageHoldingPayload, + BrokerageSnapshot, + BrokerageTransactionPayload, +) + +from .base import BaseBrokerageAdapter, BrokerageAdapterError + + +class CharlesSchwabBrokerageAdapter(BaseBrokerageAdapter): + provider = "charles-schwab" + default_base_url = "https://api.schwabapi.com/trader/v1/" + default_transaction_types = [ + "TRADE", + "DIVIDEND_OR_INTEREST", + "CASH_RECEIPT", + "CASH_DISBURSEMENT", + "ACH_RECEIPT", + "ACH_DISBURSEMENT", + "RECEIVE_AND_DELIVER", + ] + + def fetch_snapshot(self) -> BrokerageSnapshot: + account_hashes = self._fetch_account_hashes() + account_items = self._as_list( + self._request_json("/accounts", params={"fields": "positions"}) + ) + + accounts = [] + assets_by_external_id = {} + holdings = [] + transactions = [] + + for item in account_items: + account_data = item.get("securitiesAccount", item) + account_payload = self._build_account(account_data) + accounts.append(account_payload) + + for position in account_data.get("positions", []): + asset_payload = self._build_asset(position) + assets_by_external_id[asset_payload.external_id] = asset_payload + holdings.append( + self._build_holding( + account_payload=account_payload, + position=position, + asset_payload=asset_payload, + ) + ) + + account_hash = account_hashes.get(account_payload.number) + if account_hash and self.connection.get_setting("include_transactions", True): + transactions.extend( + self._fetch_transactions_for_account( + account_payload=account_payload, + account_hash=account_hash, + assets_by_external_id=assets_by_external_id, + ) + ) + + return BrokerageSnapshot( + accounts=accounts, + assets=list(assets_by_external_id.values()), + holdings=holdings, + transactions=transactions, + ) + + def _fetch_account_hashes(self) -> dict[str, str]: + payload = self._request_json("/accounts/accountNumbers") + account_numbers = self._as_list(payload) + hashes = {} + + for item in account_numbers: + account_number = item.get("accountNumber") + hash_value = item.get("hashValue") + if account_number and hash_value: + hashes[str(account_number)] = str(hash_value) + + return hashes + + def _fetch_transactions_for_account( + self, + account_payload: BrokerageAccountPayload, + account_hash: str, + assets_by_external_id: dict[str, BrokerageAssetPayload], + ) -> list[BrokerageTransactionPayload]: + transactions = [] + start_date, end_date = self._transaction_date_window() + transaction_types = self.connection.get_setting( + "transaction_types", + self.default_transaction_types, + ) + + for transaction_type in transaction_types: + payload = self._request_json( + f"/accounts/{account_hash}/transactions", + params={ + "startDate": self._format_schwab_datetime(start_date), + "endDate": self._format_schwab_datetime(end_date), + "types": transaction_type, + }, + ) + for item in self._as_list(payload): + transaction = self._build_transaction( + account_payload=account_payload, + item=item, + assets_by_external_id=assets_by_external_id, + ) + if transaction: + transactions.append(transaction) + + return transactions + + def _request_json(self, endpoint: str, params: dict | None = None): + base_url = self.connection.base_url or self.default_base_url + if not base_url.endswith("/"): + base_url = f"{base_url}/" + url = urljoin(base_url, endpoint.lstrip("/")) + if params: + url = f"{url}?{urlencode(params)}" + + request = Request(url, headers=self._headers()) + try: + with urlopen(request, timeout=30) as response: + return json.load(response) + except Exception as exc: + raise BrokerageAdapterError( + f"Unable to fetch Charles Schwab data from {url}: {exc}" + ) from exc + + def _headers(self): + access_token = self.connection.get_credential("access_token") + if not access_token: + raise BrokerageAdapterError( + "Charles Schwab ingestion requires credentials.access_token." + ) + + return { + "Accept": "application/json", + "Authorization": f"Bearer {access_token}", + } + + def _build_account(self, account_data: dict) -> BrokerageAccountPayload: + account_number = str(account_data.get("accountNumber", "")) + account_name = account_data.get("type") or account_data.get("accountType") + current_balances = account_data.get("currentBalances", {}) + projected_balances = account_data.get("projectedBalances", {}) + balance = self._first_number( + current_balances, + ["liquidationValue", "accountValue", "cashBalance"], + default=self._first_number( + projected_balances, + ["liquidationValue", "accountValue", "cashBalance"], + default=0, + ), + ) + + return BrokerageAccountPayload( + external_id=account_number, + name=f"Schwab {account_name or 'Account'} {account_number[-4:]}", + number=account_number, + balance=int(round(balance)), + ) + + def _build_asset(self, position: dict) -> BrokerageAssetPayload: + instrument = position.get("instrument", {}) + symbol = str(instrument.get("symbol") or instrument.get("cusip") or "UNKNOWN") + quantity = self._position_quantity(position) + market_value = float(position.get("marketValue") or 0) + average_price = float(position.get("averagePrice") or 0) + price = market_value / quantity if quantity else average_price + + return BrokerageAssetPayload( + external_id=symbol, + ticker=symbol, + price=price, + ) + + def _build_holding( + self, + account_payload: BrokerageAccountPayload, + position: dict, + asset_payload: BrokerageAssetPayload, + ) -> BrokerageHoldingPayload: + return BrokerageHoldingPayload( + external_id=f"{account_payload.external_id}:{asset_payload.external_id}", + account_external_id=account_payload.external_id, + asset_external_id=asset_payload.external_id, + ticker=asset_payload.ticker, + amount=self._position_quantity(position), + cost_basis=float(position.get("averagePrice") or 0), + purchase_date=datetime.now(timezone.utc), + ) + + def _build_transaction( + self, + account_payload: BrokerageAccountPayload, + item: dict, + assets_by_external_id: dict[str, BrokerageAssetPayload], + ) -> BrokerageTransactionPayload | None: + transaction_id = item.get("activityId") or item.get("transactionId") + if not transaction_id: + return None + + transaction_date = self._parse_datetime( + item.get("time") or item.get("settlementDate") or item.get("date") + ) + transfer_items = item.get("transferItems") or [] + first_transfer_item = transfer_items[0] if transfer_items else {} + transaction_type = self._map_transaction_type( + str(item.get("type", "")), + item, + first_transfer_item, + ) + instrument = first_transfer_item.get("instrument", {}) + symbol = instrument.get("symbol") or instrument.get("cusip") + amount = self._first_number( + first_transfer_item, + ["amount", "price", "cost"], + default=self._first_number(item, ["netAmount"], default=0), + ) + quantity = self._first_number( + first_transfer_item, + ["quantity"], + default=None, + ) + + if symbol and symbol not in assets_by_external_id: + assets_by_external_id[str(symbol)] = BrokerageAssetPayload( + external_id=str(symbol), + ticker=str(symbol), + price=0, + ) + + return BrokerageTransactionPayload( + external_id=str(transaction_id), + account_external_id=account_payload.external_id, + asset_external_id=str(symbol) if symbol else None, + transaction_type=transaction_type, + date=transaction_date, + value=float(amount or 0), + name=item.get("description") or item.get("subAccount") or transaction_type, + quantity=float(quantity) if quantity is not None else None, + realized_gain=self._extract_realized_gain(item), + ) + + def _transaction_date_window(self): + end_date = self._setting_datetime("transaction_end_date") or datetime.now( + timezone.utc + ) + start_date = self._setting_datetime("transaction_start_date") + if start_date: + return start_date, end_date + + days = int(self.connection.get_setting("transaction_history_days", 59)) + return end_date - timedelta(days=days), end_date + + def _setting_datetime(self, key): + value = self.connection.get_setting(key) + if not value: + return None + return self._parse_datetime(value) + + def _format_schwab_datetime(self, value: datetime): + return value.astimezone(timezone.utc).isoformat(timespec="milliseconds") + + def _parse_datetime(self, value): + if isinstance(value, datetime): + return value + if value: + return datetime.fromisoformat(str(value).replace("Z", "+00:00")) + return datetime.now(timezone.utc) + + def _as_list(self, payload): + if isinstance(payload, list): + return payload + if isinstance(payload, dict): + for key in ("accounts", "results", "data", "items"): + if isinstance(payload.get(key), list): + return payload[key] + return [] + + def _first_number(self, values: dict, keys: list[str], default=0): + for key in keys: + value = values.get(key) + if value is not None: + return float(value) + return default + + def _position_quantity(self, position: dict): + long_quantity = float(position.get("longQuantity") or 0) + short_quantity = float(position.get("shortQuantity") or 0) + return long_quantity - short_quantity + + def _map_transaction_type(self, schwab_type: str, item: dict, transfer_item: dict): + normalized = schwab_type.upper() + if normalized == "DIVIDEND_OR_INTEREST": + return "dividend" + if normalized in {"ACH_RECEIPT", "CASH_RECEIPT", "WIRE_IN"}: + return "deposit" + if normalized in {"ACH_DISBURSEMENT", "CASH_DISBURSEMENT", "WIRE_OUT"}: + return "withdrawal" + if normalized == "TRADE": + trade_text = " ".join( + str(value) + for value in [ + item.get("description"), + item.get("transactionSubType"), + transfer_item.get("instruction"), + ] + if value + ).upper() + if "SELL" in trade_text or "SOLD" in trade_text: + return "sell" + if "BUY" in trade_text or "BOUGHT" in trade_text: + return "buy" + return "trade" + return normalized.lower() + + def _extract_realized_gain(self, item: dict): + for key in ("realizedGain", "realizedGainLoss", "gainLoss"): + value = item.get(key) + if value is not None: + return float(value) + return None diff --git a/overview/brokerages/registry.py b/overview/brokerages/registry.py index a709417..79e9ba7 100644 --- a/overview/brokerages/registry.py +++ b/overview/brokerages/registry.py @@ -4,11 +4,13 @@ from django.utils.module_loading import import_string from overview.models import BrokerageConnection from .base import BaseBrokerageAdapter, BrokerageAdapterError +from .charles_schwab import CharlesSchwabBrokerageAdapter from .random import RandomBrokerageAdapter from .rest import GenericRestBrokerageAdapter _DEFAULT_ADAPTERS = { + CharlesSchwabBrokerageAdapter.provider: CharlesSchwabBrokerageAdapter, GenericRestBrokerageAdapter.provider: GenericRestBrokerageAdapter, RandomBrokerageAdapter.provider: RandomBrokerageAdapter, } diff --git a/overview/tests.py b/overview/tests.py index 97f7057..451dce9 100644 --- a/overview/tests.py +++ b/overview/tests.py @@ -14,6 +14,7 @@ from .brokerage_types import ( BrokerageTransactionPayload, ) from .brokerages.base import BaseBrokerageAdapter, BrokerageAdapterError +from .brokerages.charles_schwab import CharlesSchwabBrokerageAdapter from .brokerages.random import RandomBrokerageAdapter from .brokerages.registry import register_brokerage_adapter from .models import ( @@ -55,6 +56,19 @@ class ExplodingBrokerageAdapter(BaseBrokerageAdapter): raise RuntimeError("brokerage unavailable") +class FakeCharlesSchwabBrokerageAdapter(CharlesSchwabBrokerageAdapter): + def __init__(self, connection, responses): + super().__init__(connection) + self.responses = responses + self.requests = [] + + def _request_json(self, endpoint: str, params: dict | None = None): + self.requests.append((endpoint, params or {})) + if endpoint.startswith("/accounts/") and endpoint.endswith("/transactions"): + return self.responses["transactions"] + return self.responses[endpoint] + + class BrokerageIngestionServiceTests(TestCase): def setUp(self): register_brokerage_adapter(FakeBrokerageAdapter.provider, FakeBrokerageAdapter) @@ -366,6 +380,94 @@ class BrokerageIngestionServiceTests(TestCase): 18, ) + def test_charles_schwab_adapter_maps_accounts_positions_and_transactions(self): + connection = BrokerageConnection.objects.create( + provider=CharlesSchwabBrokerageAdapter.provider, + name="Schwab", + base_url="https://api.schwabapi.com/trader/v1/", + credentials={"access_token": "token-1"}, + settings={ + "transaction_start_date": "2026-01-01T00:00:00+00:00", + "transaction_end_date": "2026-01-31T00:00:00+00:00", + "transaction_types": ["TRADE"], + }, + ) + adapter = FakeCharlesSchwabBrokerageAdapter( + connection, + responses={ + "/accounts/accountNumbers": [ + { + "accountNumber": "123456789", + "hashValue": "encrypted-account-1", + } + ], + "/accounts": [ + { + "securitiesAccount": { + "type": "MARGIN", + "accountNumber": "123456789", + "currentBalances": { + "liquidationValue": 25000.49, + }, + "positions": [ + { + "longQuantity": 10, + "averagePrice": 150.25, + "marketValue": 2000, + "instrument": { + "symbol": "AAPL", + "cusip": "037833100", + }, + } + ], + } + } + ], + "transactions": [ + { + "activityId": 123, + "type": "TRADE", + "description": "SELL AAPL", + "time": "2026-01-15T15:30:00+00:00", + "netAmount": -2100, + "realizedGain": 300.5, + "transferItems": [ + { + "quantity": 10, + "amount": -2100, + "instruction": "SELL", + "instrument": { + "symbol": "AAPL", + }, + } + ], + } + ], + }, + ) + + snapshot = adapter.fetch_snapshot() + + self.assertEqual(snapshot.accounts[0].external_id, "123456789") + self.assertEqual(snapshot.accounts[0].balance, 25000) + self.assertEqual(snapshot.assets[0].ticker, "AAPL") + self.assertEqual(snapshot.assets[0].price, 200) + self.assertEqual(snapshot.holdings[0].amount, 10) + self.assertEqual(snapshot.transactions[0].transaction_type, "sell") + self.assertEqual(snapshot.transactions[0].realized_gain, 300.5) + self.assertEqual(snapshot.transactions[0].quantity, 10) + self.assertIn( + ( + "/accounts/encrypted-account-1/transactions", + { + "startDate": "2026-01-01T00:00:00.000+00:00", + "endDate": "2026-01-31T00:00:00.000+00:00", + "types": "TRADE", + }, + ), + adapter.requests, + ) + def test_sync_command_accepts_provider_identifier(self): random_connection = BrokerageConnection.objects.create( provider=RandomBrokerageAdapter.provider,