06/02 Integrate Schwab
This commit is contained in:
@@ -0,0 +1,315 @@
|
||||
"""
|
||||
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 → list accounts
|
||||
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',
|
||||
}
|
||||
|
||||
|
||||
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',
|
||||
})
|
||||
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_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_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 timezone offset
|
||||
raw_time = schwab_txn.get('time', '')
|
||||
try:
|
||||
txn_date = datetime.fromisoformat(raw_time.replace('Z', '+00:00')).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
|
||||
Reference in New Issue
Block a user