456 lines
16 KiB
Python
456 lines
16 KiB
Python
"""
|
|
Schwab Developer API Service — OAuth 2.0 + Trader API
|
|
|
|
Auth flow:
|
|
1. Redirect user to SCHWAB_AUTH_URL with client_id + redirect_uri + state
|
|
2. Schwab calls back with ?code=...&state=...
|
|
3. Exchange code for access_token + refresh_token (Basic Auth: client_id:client_secret)
|
|
4. Access token expires in 30 min — auto-refresh via refresh_token (valid 7 days)
|
|
|
|
Endpoints used:
|
|
GET /trader/v1/accounts/accountNumbers → {accountNumber: hashValue} map
|
|
GET /trader/v1/accounts?fields=positions → list accounts with balances + positions
|
|
GET /trader/v1/accounts/{hash}?fields=positions → single account with balance + positions
|
|
GET /trader/v1/accounts/{hash}/transactions?startDate&endDate → transactions
|
|
"""
|
|
|
|
import base64
|
|
import logging
|
|
import os
|
|
from datetime import date, datetime, timedelta
|
|
|
|
import requests
|
|
from flask import current_app
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
SCHWAB_AUTH_URL = 'https://api.schwabapi.com/v1/oauth/authorize'
|
|
SCHWAB_TOKEN_URL = 'https://api.schwabapi.com/v1/oauth/token'
|
|
SCHWAB_BASE = 'https://api.schwabapi.com'
|
|
|
|
# Schwab transaction type → PFM category name
|
|
CATEGORY_MAP = {
|
|
'DIVIDEND_OR_INTEREST': 'Investment',
|
|
'TRADE': 'Investment',
|
|
'BUY': 'Investment',
|
|
'SELL': 'Investment',
|
|
'ACH_RECEIPT': 'Other Income',
|
|
'ACH_DISBURSEMENT': 'Other',
|
|
'WIRE_IN': 'Other Income',
|
|
'WIRE_OUT': 'Other',
|
|
'CASH_RECEIPT': 'Other Income',
|
|
'CASH_DISBURSEMENT': 'Other',
|
|
'ELECTRONIC_FUND': 'Other',
|
|
'RECEIVE_AND_DELIVER': 'Investment',
|
|
'TRANSFER_OF_ACCOUNT_IN': 'Other Income',
|
|
'TRANSFER_OF_ACCOUNT_OUT': 'Other',
|
|
'JOURNAL': 'Other',
|
|
'PASS_THROUGH_CHARGE': 'Other',
|
|
'PASS_THROUGH_REBATE': 'Other Income',
|
|
'TRUST_FEES': 'Other',
|
|
'MEMORIAL': 'Other',
|
|
}
|
|
|
|
# Schwab account type → PFM account type
|
|
ACCOUNT_TYPE_MAP = {
|
|
'CASH': 'checking',
|
|
'MARGIN': 'investment',
|
|
}
|
|
|
|
# Schwab instrument asset type → PFM investment asset type
|
|
ASSET_TYPE_MAP = {
|
|
'EQUITY': 'stock',
|
|
'ETF': 'etf',
|
|
'MUTUAL_FUND': 'etf',
|
|
'COLLECTIVE_INVESTMENT': 'etf',
|
|
'INDEX': 'etf',
|
|
'FIXED_INCOME': 'bond',
|
|
'BOND': 'bond',
|
|
'CASH_EQUIVALENT': 'cash',
|
|
'CURRENCY': 'cash',
|
|
'OPTION': 'other',
|
|
'FUTURE': 'other',
|
|
}
|
|
|
|
|
|
def get_auth_url():
|
|
cfg = current_app.config
|
|
client_id = cfg['SCHWAB_CLIENT_ID']
|
|
redirect_uri = cfg['SCHWAB_REDIRECT_URI']
|
|
import urllib.parse, secrets
|
|
state = secrets.token_urlsafe(16)
|
|
params = urllib.parse.urlencode({
|
|
'client_id': client_id,
|
|
'redirect_uri': redirect_uri,
|
|
'response_type': 'code',
|
|
'scope': 'readonly',
|
|
'state': state,
|
|
})
|
|
return f'{SCHWAB_AUTH_URL}?{params}', state
|
|
|
|
|
|
def _basic_auth_header():
|
|
cfg = current_app.config
|
|
creds = f"{cfg['SCHWAB_CLIENT_ID']}:{cfg['SCHWAB_CLIENT_SECRET']}"
|
|
return 'Basic ' + base64.b64encode(creds.encode()).decode()
|
|
|
|
|
|
def exchange_code(code):
|
|
"""Exchange authorization code for tokens. Returns token dict."""
|
|
cfg = current_app.config
|
|
resp = requests.post(
|
|
SCHWAB_TOKEN_URL,
|
|
headers={
|
|
'Authorization': _basic_auth_header(),
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
},
|
|
data={
|
|
'grant_type': 'authorization_code',
|
|
'code': code,
|
|
'redirect_uri': cfg['SCHWAB_REDIRECT_URI'],
|
|
},
|
|
timeout=15,
|
|
)
|
|
_raise_for_status(resp, 'exchange_code')
|
|
return resp.json()
|
|
|
|
|
|
def refresh_tokens(connection):
|
|
"""
|
|
Refresh access token using stored refresh_token.
|
|
Updates connection object in-place and commits to DB.
|
|
Raises on failure.
|
|
"""
|
|
resp = requests.post(
|
|
SCHWAB_TOKEN_URL,
|
|
headers={
|
|
'Authorization': _basic_auth_header(),
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
},
|
|
data={
|
|
'grant_type': 'refresh_token',
|
|
'refresh_token': connection.refresh_token,
|
|
},
|
|
timeout=15,
|
|
)
|
|
_raise_for_status(resp, 'refresh_tokens')
|
|
data = resp.json()
|
|
_apply_token_data(connection, data)
|
|
from app.extensions import db
|
|
db.session.commit()
|
|
log.info('[schwab] tokens refreshed for connection id=%s', connection.id)
|
|
return connection
|
|
|
|
|
|
def _apply_token_data(connection, data):
|
|
"""Write token fields from token response onto connection model."""
|
|
connection.access_token = data['access_token']
|
|
connection.refresh_token = data.get('refresh_token', connection.refresh_token)
|
|
expires_in = int(data.get('expires_in', 1800))
|
|
connection.token_expires_at = datetime.utcnow() + timedelta(seconds=expires_in - 60)
|
|
|
|
|
|
def _ensure_fresh(connection):
|
|
"""Auto-refresh access token if within 60 seconds of expiry."""
|
|
if connection.token_is_expired:
|
|
log.info('[schwab] access token expired — refreshing')
|
|
refresh_tokens(connection)
|
|
|
|
|
|
def _authed_get(connection, path, params=None):
|
|
"""GET request with auto token refresh. Returns parsed JSON."""
|
|
_ensure_fresh(connection)
|
|
url = SCHWAB_BASE + path
|
|
resp = requests.get(
|
|
url,
|
|
headers={'Authorization': f'Bearer {connection.access_token}'},
|
|
params=params or {},
|
|
timeout=30,
|
|
)
|
|
_raise_for_status(resp, f'GET {path}')
|
|
return resp.json()
|
|
|
|
|
|
def _raise_for_status(resp, context=''):
|
|
if not resp.ok:
|
|
log.error('[schwab] API error (%s) — status=%s body=%r',
|
|
context, resp.status_code, resp.text[:2000])
|
|
resp.raise_for_status()
|
|
|
|
|
|
# ── Public API helpers ────────────────────────────────────────────────────────
|
|
|
|
def get_account_number_hashes(connection):
|
|
"""
|
|
Return {accountNumber: hashValue} mapping.
|
|
Schwab requires the hashValue (encrypted account number) in all endpoint paths.
|
|
"""
|
|
data = _authed_get(connection, '/trader/v1/accounts/accountNumbers')
|
|
mapping = {item['accountNumber']: item['hashValue'] for item in data}
|
|
log.info('[schwab] get_account_number_hashes: %d account(s)', len(mapping))
|
|
return mapping
|
|
|
|
|
|
def get_accounts(connection):
|
|
"""Return list of Schwab account dicts."""
|
|
data = _authed_get(connection, '/trader/v1/accounts', params={'fields': 'positions'})
|
|
log.info('[schwab] get_accounts: returned %d account(s)', len(data))
|
|
return data
|
|
|
|
|
|
def get_account(connection, account_hash):
|
|
"""Fetch a single account's balance and positions by its hash."""
|
|
data = _authed_get(connection, f'/trader/v1/accounts/{account_hash}',
|
|
params={'fields': 'positions'})
|
|
log.info('[schwab] get_account: hash=%s…', account_hash[:8])
|
|
return data
|
|
|
|
|
|
def sync_account_snapshot(schwab_account):
|
|
"""
|
|
Pull live balance and equity positions for one Schwab account and write to PFM.
|
|
|
|
Balance: sets the linked PFM account balance to Schwab's liquidationValue
|
|
(total portfolio value = cash + market value of all holdings).
|
|
Positions: upserts Investment records for every long equity/ETF/fund/bond
|
|
position; updates shares, avg cost, and current price.
|
|
|
|
Returns (balance_updated: bool, positions_synced: int).
|
|
"""
|
|
from app.extensions import db
|
|
from app.models.investment import Investment
|
|
|
|
connection = schwab_account.connection
|
|
data = get_account(connection, schwab_account.account_hash)
|
|
sec = data.get('securitiesAccount', {})
|
|
|
|
# ── 1. Balance ────────────────────────────────────────────────────────────
|
|
balance_updated = False
|
|
if schwab_account.pfm_account:
|
|
balances = sec.get('currentBalances', {})
|
|
liq_value = float(
|
|
balances.get('liquidationValue') or
|
|
balances.get('cashBalance') or 0
|
|
)
|
|
schwab_account.pfm_account.balance = liq_value
|
|
balance_updated = True
|
|
log.info('[schwab] balance set to %.2f for %s',
|
|
liq_value, schwab_account.account_name)
|
|
|
|
# ── 2. Positions ──────────────────────────────────────────────────────────
|
|
positions_synced = 0
|
|
raw_positions = sec.get('positions') or [] # guard: API may send null
|
|
log.info('[schwab] %s has %d position(s) in API response',
|
|
schwab_account.account_name, len(raw_positions))
|
|
|
|
for pos in raw_positions:
|
|
instrument = pos.get('instrument') or {}
|
|
asset_key = instrument.get('assetType', '')
|
|
symbol = (instrument.get('symbol') or '').upper().strip()
|
|
# Use longQuantity; fall back to settledLongQuantity for positions that
|
|
# were just purchased and haven't fully settled yet (T+1 / T+2).
|
|
long_qty = float(pos.get('longQuantity') or
|
|
pos.get('settledLongQuantity') or 0)
|
|
|
|
pfm_type = ASSET_TYPE_MAP.get(asset_key)
|
|
log.debug('[schwab] position: symbol=%s assetType=%s longQty=%s pfm_type=%s',
|
|
symbol, asset_key, long_qty, pfm_type)
|
|
|
|
# Skip empty symbols and zero-quantity positions
|
|
if not symbol or long_qty <= 0:
|
|
log.debug('[schwab] skipping %s — qty=%s symbol=%r', asset_key, long_qty, symbol)
|
|
continue
|
|
# Fall back to 'other' if the asset type isn't in our map
|
|
if not pfm_type:
|
|
pfm_type = 'other'
|
|
|
|
avg_price = float(pos.get('averagePrice') or pos.get('averageLongPrice') or 0)
|
|
market_value = float(pos.get('marketValue') or 0)
|
|
cur_price = round(market_value / long_qty, 4) if long_qty > 0 else avg_price
|
|
|
|
pfm_acct_id = schwab_account.pfm_account_id
|
|
|
|
# Match on (ticker, account_id) so the same ticker in different accounts
|
|
# (e.g. AAPL in Individual vs Roth IRA) remains separate.
|
|
inv = Investment.query.filter_by(
|
|
ticker=symbol, account_id=pfm_acct_id, is_active=True
|
|
).first()
|
|
if inv:
|
|
inv.shares = long_qty
|
|
if avg_price > 0:
|
|
inv.avg_cost_basis = avg_price
|
|
inv.current_price = cur_price
|
|
inv.last_price_update = datetime.utcnow()
|
|
else:
|
|
name = (instrument.get('description') or symbol).strip()
|
|
inv = Investment(
|
|
account_id = pfm_acct_id,
|
|
asset_name = name,
|
|
ticker = symbol,
|
|
asset_type = pfm_type,
|
|
shares = long_qty,
|
|
avg_cost_basis = avg_price,
|
|
current_price = cur_price,
|
|
last_price_update = datetime.utcnow(),
|
|
notes = 'Imported from Schwab',
|
|
)
|
|
db.session.add(inv)
|
|
|
|
positions_synced += 1
|
|
|
|
db.session.commit()
|
|
log.info('[schwab] snapshot done for %s: positions=%d',
|
|
schwab_account.account_name, positions_synced)
|
|
return balance_updated, positions_synced
|
|
|
|
|
|
def get_transactions(connection, account_hash, start_date, end_date):
|
|
"""
|
|
Fetch transactions for one account.
|
|
start_date / end_date: date objects or ISO strings.
|
|
Returns list of transaction dicts.
|
|
"""
|
|
def _iso(d):
|
|
if hasattr(d, 'strftime'):
|
|
return d.strftime('%Y-%m-%dT00:00:00.000Z')
|
|
return d
|
|
|
|
params = {
|
|
'startDate': _iso(start_date),
|
|
'endDate': _iso(end_date),
|
|
}
|
|
data = _authed_get(connection, f'/trader/v1/accounts/{account_hash}/transactions', params)
|
|
log.info('[schwab] get_transactions: account=%s returned %d txn(s)',
|
|
account_hash[:8] + '…', len(data))
|
|
return data
|
|
|
|
|
|
def build_category_map():
|
|
from app.models.category import Category
|
|
cats = Category.query.filter_by(is_active=True).all()
|
|
return {c.name: c.id for c in cats}
|
|
|
|
|
|
def parse_transaction(schwab_txn, pfm_account_id, cat_id_map):
|
|
"""
|
|
Convert a Schwab transaction dict to a PFM-ready dict.
|
|
|
|
Schwab netAmount convention:
|
|
positive → money came INTO the account (income)
|
|
negative → money LEFT the account (expense)
|
|
"""
|
|
net = float(schwab_txn.get('netAmount', 0))
|
|
if net >= 0:
|
|
txn_type = 'income'
|
|
amount = net
|
|
else:
|
|
txn_type = 'expense'
|
|
amount = abs(net)
|
|
|
|
# Prefer description, fall back to type
|
|
description = (schwab_txn.get('description') or
|
|
schwab_txn.get('type', 'Schwab transaction')).strip()
|
|
|
|
schwab_type = schwab_txn.get('type', '')
|
|
cat_name = CATEGORY_MAP.get(schwab_type, 'Other')
|
|
category_id = cat_id_map.get(cat_name)
|
|
|
|
# Parse date — Schwab uses ISO-8601 with various timezone offset forms
|
|
# e.g. "2024-01-05T18:45:45+0000" or "2024-01-05T18:45:45Z"
|
|
raw_time = schwab_txn.get('time', '') or ''
|
|
try:
|
|
normalized = raw_time.replace('Z', '+00:00')
|
|
# Normalise +0000 → +00:00 so fromisoformat accepts it on all Python versions
|
|
import re as _re
|
|
normalized = _re.sub(r'([+-]\d{2})(\d{2})$', r'\1:\2', normalized)
|
|
txn_date = datetime.fromisoformat(normalized).date()
|
|
except (ValueError, AttributeError):
|
|
txn_date = date.today()
|
|
|
|
activity_id = str(schwab_txn.get('activityId', ''))
|
|
|
|
return {
|
|
'schwab_id': activity_id,
|
|
'date': txn_date,
|
|
'transaction_type': txn_type,
|
|
'amount': amount,
|
|
'description': description,
|
|
'account_id': pfm_account_id,
|
|
'category_id': category_id,
|
|
'notes': f'Schwab:{activity_id}',
|
|
'schwab_type': schwab_type,
|
|
}
|
|
|
|
|
|
def sync_preview(schwab_account, days_back=90):
|
|
"""
|
|
Fetch and parse transactions for a SchwabAccount.
|
|
Returns list of parsed dicts — does NOT write to DB.
|
|
"""
|
|
connection = schwab_account.connection
|
|
today = date.today()
|
|
|
|
if schwab_account.last_sync_date:
|
|
start = schwab_account.last_sync_date - timedelta(days=7)
|
|
else:
|
|
start = today - timedelta(days=days_back)
|
|
|
|
raw_txns = get_transactions(
|
|
connection,
|
|
schwab_account.account_hash,
|
|
start_date=start,
|
|
end_date=today,
|
|
)
|
|
|
|
cat_map = build_category_map()
|
|
return [
|
|
parse_transaction(t, schwab_account.pfm_account_id, cat_map)
|
|
for t in raw_txns
|
|
if float(t.get('netAmount', 0)) != 0 # skip zero-amount entries
|
|
]
|
|
|
|
|
|
def import_transactions(parsed_txns, schwab_account):
|
|
"""
|
|
Import parsed transactions. Skips duplicates via Schwab:<activityId> in notes.
|
|
Returns (imported_count, skipped_count).
|
|
"""
|
|
from app.extensions import db
|
|
from app.models.transaction import Transaction
|
|
from app.services.account_service import calc_balance
|
|
|
|
imported = skipped = 0
|
|
affected = set()
|
|
|
|
for p in parsed_txns:
|
|
sid = p['schwab_id']
|
|
if Transaction.query.filter(Transaction.notes.like(f'%Schwab:{sid}%')).first():
|
|
skipped += 1
|
|
continue
|
|
|
|
db.session.add(Transaction(
|
|
account_id = p['account_id'],
|
|
category_id = p.get('category_id'),
|
|
transaction_type = p['transaction_type'],
|
|
amount = p['amount'],
|
|
description = p['description'],
|
|
date = p['date'],
|
|
notes = p['notes'],
|
|
))
|
|
if p['account_id']:
|
|
affected.add(p['account_id'])
|
|
imported += 1
|
|
|
|
db.session.commit()
|
|
|
|
schwab_account.last_sync_date = date.today()
|
|
schwab_account.connection.last_synced_at = datetime.utcnow()
|
|
if parsed_txns:
|
|
schwab_account.last_schwab_txn_id = parsed_txns[0]['schwab_id']
|
|
db.session.commit()
|
|
|
|
for acct_id in affected:
|
|
calc_balance(acct_id)
|
|
|
|
return imported, skipped
|