broker-sync/broker_sync/providers/parsers/schwab.py

"""Schwab workplace-RSU email parser.

Schwab Stock Plan Services sends a "Your trade was executed" email for
each sell-to-cover trade (and any user-initiated trade) on the workplace
account. The body has five `<td class="dark-background-body" align="right">`
cells holding date / direction / quantity / ticker / price.

It does NOT email vest-release / Release Confirmation messages to the
employee address for this account (verified against 4 years of inbox
history, 2022-2026). The vest itself is invisible to IMAP.

Same-day-sell synthesis: Meta RSUs vest and are sold the same day at
the same FMV (verified across 14 historical vests). When a SELL email
is parsed AND its trade date is on or after `VEST_INFER_FROM_DATE`,
we ALSO emit a paired BUY representing the underlying vest event —
same date, same quantity, same price. The date boundary stops this
back-filling historical vests that already have csv-sourced BUY rows
in Wealthfolio (which would duplicate at chart-level despite distinct
external_ids).

On any parse failure we return an empty list — an unparseable email
shouldn't crash the IMAP batch.
"""
from __future__ import annotations

import logging
import os
from datetime import date, datetime
from decimal import Decimal, InvalidOperation

from bs4 import BeautifulSoup
from dateutil import parser as dateparser

from broker_sync.models import AccountType, Activity, ActivityType

log = logging.getLogger(__name__)

_ACCOUNT_ID = "schwab-workplace"
_DEFAULT_CURRENCY = "USD"

# Inferred-BUY synthesis boundary. SELL emails on or after this date
# emit a paired BUY for the underlying vest; earlier ones do not (they
# already have csv-sourced BUYs in Wealthfolio from the one-shot
# historical backfill, last vest 2026-02-18). Override at runtime with
# the env var if a different cutover is needed. ISO-8601 yyyy-mm-dd.
_DEFAULT_VEST_INFER_FROM = "2026-04-01"


def _vest_infer_from() -> date:
    raw = os.environ.get("SCHWAB_VEST_INFER_FROM_DATE", _DEFAULT_VEST_INFER_FROM).strip()
    try:
        return datetime.strptime(raw, "%Y-%m-%d").date()
    except ValueError:
        log.warning(
            "SCHWAB_VEST_INFER_FROM_DATE=%r is not yyyy-mm-dd; using default %s",
            raw, _DEFAULT_VEST_INFER_FROM,
        )
        return datetime.strptime(_DEFAULT_VEST_INFER_FROM, "%Y-%m-%d").date()


def parse_schwab_email(raw_html: str) -> list[Activity]:
    """Return Activities for a Schwab trade-executed email.

    Returns: empty list on parse failure; one Activity for a BUY-direction
    email (rare — the workplace account is essentially sell-only); for a
    SELL email, returns [SELL] plus an inferred paired BUY (=vest event)
    when the trade date is on or after the synthesis-boundary date.
    """
    try:
        soup = BeautifulSoup(raw_html, "html.parser")
        cells = [
            td.get_text(strip=True) for td in soup.find_all("td", {
                "class": "dark-background-body",
                "align": "right"
            })
        ]
        if len(cells) < 5:
            return []

        date_txt, direction_txt, qty_txt, ticker, price_txt = cells[:5]
        trade_date = dateparser.parse(date_txt)
        direction = (ActivityType.SELL
                     if direction_txt.strip().lower() == "sold" else ActivityType.BUY)
        quantity = Decimal(qty_txt.replace(",", "").strip())
        price_clean = price_txt
        for sign in ("$", "£", "€", "USD", "GBP", "EUR"):
            price_clean = price_clean.replace(sign, "")
        unit_price = Decimal(price_clean.replace(",", "").strip())
        ticker_clean = ticker.strip()

        external_id = (f"schwab:{trade_date.date().isoformat()}:{ticker_clean}:"
                       f"{direction.value}:{quantity}")
        primary = Activity(
            external_id=external_id,
            account_id=_ACCOUNT_ID,
            account_type=AccountType.GIA,
            date=trade_date,
            activity_type=direction,
            symbol=ticker_clean,
            quantity=quantity,
            unit_price=unit_price,
            currency=_DEFAULT_CURRENCY,
            notes=f"schwab-email:{direction_txt}",
        )

        if direction is not ActivityType.SELL or trade_date.date() < _vest_infer_from():
            return [primary]

        inferred_buy = Activity(
            external_id=(f"schwab:vest:{trade_date.date().isoformat()}:"
                         f"{ticker_clean}:BUY:{quantity}"),
            account_id=_ACCOUNT_ID,
            account_type=AccountType.GIA,
            date=trade_date,
            activity_type=ActivityType.BUY,
            symbol=ticker_clean,
            quantity=quantity,
            unit_price=unit_price,
            currency=_DEFAULT_CURRENCY,
            notes=(f"schwab-vest-inferred-from-same-day-sell | "
                   f"paired_sell_external_id={external_id}"),
        )
        return [inferred_buy, primary]
    except (ValueError, InvalidOperation, IndexError, AttributeError):
        return []