Implementation Guide: Identify tax-loss harvesting opportunities across client accounts seasonally

Get-WmiObject -Class Win32_Product | Select-Object Name, Version | Export-Csv C:\Audit\InstalledSoftware.csv
Invoke-WebRequest -Uri 'https://speed.cloudflare.com' -UseBasicParsing | Out-Null
(Get-ItemProperty 'HKLM:\SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\chrome.exe').'(default)' | ForEach-Object { & $_ --version }

config system interface
  edit port1
    set mode dhcp
    set allowaccess ping https ssh
    set alias WAN1
  next
end

config system interface
  edit vlan10
    set vdom root
    set ip 10.10.10.1 255.255.255.0
    set allowaccess ping https
    set interface port3
    set vlanid 10
    set alias ADVISOR_LAN
  next
end

config system fortiguard
  set port 443
  set protocol https
end

config ips sensor
  edit financial-grade
    config entries
      edit 1
        set status enable
        set log-packet enable
        set severity critical high medium
      next
    end
  next
end

msiexec /i CrowdStrike-Falcon-Sensor.msi /quiet CID=<YOUR_CUSTOMER_ID> ProvNoWait=1

capture@<firmname>.smarsh.cloud

# Orion SAML SSO Configuration
SAML SSO URL: https://login.orion.com/sso/saml
Entity ID: Provided by Orion during onboarding
Claim mappings: email, first_name, last_name, advisor_id

SPY (S&P 500 ETF)      -> IVV or VOO
QQQ (Nasdaq 100)       -> QQQM or VGT
VTI (Total Market)     -> ITOT or SCHB
VXUS (Intl Developed)  -> IXUS or SPDW
BND (Total Bond)       -> AGG or SCHZ
VNQ (Real Estate)      -> SCHH or IYR

az login
az group create --name rg-tlh-advisor --location eastus
az storage account create --name stgtlhadvisor --resource-group rg-tlh-advisor --sku Standard_LRS
az functionapp create --name func-tlh-scanner --resource-group rg-tlh-advisor --storage-account stgtlhadvisor --consumption-plan-location eastus --runtime python --runtime-version 3.11 --functions-version 4

cd tlh-scanner-function
func azure functionapp publish func-tlh-scanner

import requests
import json
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.html import MIMEHTML
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Optional
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('tlh_scorer')

# ============================================================
# CONFIGURATION - Update these values per client deployment
# ============================================================
ORION_API_BASE = 'https://api.orion.com/api/v1'
ORION_API_KEY = 'YOUR_ORION_API_KEY'  # Store in Azure Key Vault or env var
SMTP_SERVER = 'smtp.office365.com'
SMTP_PORT = 587
SMTP_USER = 'alerts@advisorfirm.com'
SMTP_PASS = 'USE_APP_PASSWORD'  # Store in Key Vault
ALERT_RECIPIENTS = ['lead.advisor@firm.com', 'ops@firm.com']
CCO_EMAIL = 'cco@firm.com'

# TLH Policy Parameters (configured per firm)
MIN_LOSS_THRESHOLD = 100  # Minimum $ loss to consider harvesting
MIN_HOLDING_PERIOD_DAYS = 31  # Must hold >30 days (wash-sale safe)
SHORT_TERM_GAIN_TOLERANCE = 0  # Don't create short-term gains
MARKET_CORRECTION_THRESHOLD = -0.05  # 5% S&P decline triggers alert


@dataclass
class TaxProfile:
    household_id: str
    client_name: str
    marginal_rate: float  # e.g., 0.37 for 37%
    ltcg_rate: float      # e.g., 0.20 for 20%
    stcg_rate: float      # e.g., 0.37 (same as marginal)
    capital_loss_carryforward: float  # $ amount from prior years
    amt_flag: bool
    state_rate: float     # State income tax rate


@dataclass
class TLHOpportunity:
    household_id: str
    client_name: str
    account_id: str
    account_name: str
    security_ticker: str
    security_name: str
    quantity: float
    cost_basis: float
    current_value: float
    unrealized_loss: float
    loss_type: str  # 'short_term' or 'long_term'
    holding_period_days: int
    substitute_ticker: str
    substitute_name: str
    wash_sale_risk: bool
    wash_sale_detail: Optional[str]
    tax_benefit_estimate: float
    priority_score: float


# ============================================================
# SUBSTITUTE SECURITY MAPPING TABLE
# Firm-specific: must be approved by lead advisor and CCO
# ============================================================
SUBSTITUTE_MAP = {
    'SPY':  {'sub': 'IVV',  'name': 'iShares Core S&P 500 ETF'},
    'IVV':  {'sub': 'VOO',  'name': 'Vanguard S&P 500 ETF'},
    'VOO':  {'sub': 'SPY',  'name': 'SPDR S&P 500 ETF Trust'},
    'QQQ':  {'sub': 'QQQM', 'name': 'Invesco NASDAQ 100 ETF'},
    'QQQM': {'sub': 'QQQ',  'name': 'Invesco QQQ Trust'},
    'VTI':  {'sub': 'ITOT', 'name': 'iShares Core S&P Total US Stock Market ETF'},
    'ITOT': {'sub': 'SCHB', 'name': 'Schwab US Broad Market ETF'},
    'SCHB': {'sub': 'VTI',  'name': 'Vanguard Total Stock Market ETF'},
    'VXUS': {'sub': 'IXUS', 'name': 'iShares Core MSCI Total Intl Stock ETF'},
    'IXUS': {'sub': 'SPDW', 'name': 'SPDR Portfolio Developed World ex-US ETF'},
    'VWO':  {'sub': 'IEMG', 'name': 'iShares Core MSCI Emerging Markets ETF'},
    'BND':  {'sub': 'AGG',  'name': 'iShares Core US Aggregate Bond ETF'},
    'AGG':  {'sub': 'SCHZ', 'name': 'Schwab US Aggregate Bond ETF'},
    'VNQ':  {'sub': 'SCHH', 'name': 'Schwab US REIT ETF'},
    'SCHH': {'sub': 'IYR',  'name': 'iShares US Real Estate ETF'},
    'VEA':  {'sub': 'SPDW', 'name': 'SPDR Portfolio Developed World ex-US ETF'},
    'EFA':  {'sub': 'VEA',  'name': 'Vanguard FTSE Developed Markets ETF'},
    'VIG':  {'sub': 'DGRO', 'name': 'iShares Core Dividend Growth ETF'},
    'SCHD': {'sub': 'VIG',  'name': 'Vanguard Dividend Appreciation ETF'},
}


# ============================================================
# TAX PROFILE STORE
# Populated from Holistiplan export + Redtail custom fields
# In production, this reads from a database or API
# ============================================================
def load_tax_profiles() -> dict:
    """Load tax profiles from Redtail custom fields or local CSV.
    Returns dict keyed by household_id."""
    # Example: read from CSV exported from Holistiplan/Redtail
    import csv
    profiles = {}
    try:
        with open('/data/tax_profiles.csv', 'r') as f:
            reader = csv.DictReader(f)
            for row in reader:
                profiles[row['household_id']] = TaxProfile(
                    household_id=row['household_id'],
                    client_name=row['client_name'],
                    marginal_rate=float(row['marginal_rate']),
                    ltcg_rate=float(row['ltcg_rate']),
                    stcg_rate=float(row['stcg_rate']),
                    capital_loss_carryforward=float(row.get('loss_carryforward', 0)),
                    amt_flag=row.get('amt_flag', 'false').lower() == 'true',
                    state_rate=float(row.get('state_rate', 0))
                )
    except FileNotFoundError:
        logger.warning('Tax profiles CSV not found. Using default rates.')
    return profiles


def get_orion_holdings(api_base: str, api_key: str) -> list:
    """Fetch all holdings with unrealized gain/loss data from Orion API."""
    headers = {'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'}
    holdings = []
    page = 1
    while True:
        resp = requests.get(
            f'{api_base}/portfolio/holdings',
            headers=headers,
            params={
                'page': page,
                'pageSize': 500,
                'includeUnrealizedGL': True,
                'includeTaxLots': True,
                'accountTypes': 'taxable'  # Only taxable accounts
            }
        )
        resp.raise_for_status()
        data = resp.json()
        holdings.extend(data.get('results', []))
        if page >= data.get('totalPages', 1):
            break
        page += 1
    return holdings


def check_wash_sale_risk(household_id: str, ticker: str, api_base: str, api_key: str) -> tuple:
    """Check if selling this ticker creates wash-sale risk due to recent
    purchases of the same or substantially identical security in any
    account within the household (including IRAs/401k).
    Returns (is_risky: bool, detail: str)"""
    headers = {'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'}
    
    # Get all transactions for this household in the wash-sale window
    window_start = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
    window_end = (datetime.now() + timedelta(days=30)).strftime('%Y-%m-%d')
    
    resp = requests.get(
        f'{api_base}/portfolio/transactions',
        headers=headers,
        params={
            'householdId': household_id,
            'ticker': ticker,
            'transactionType': 'buy',
            'startDate': window_start,
            'endDate': window_end,
            'includeAllAccountTypes': True  # Include IRAs, 401(k)s
        }
    )
    resp.raise_for_status()
    transactions = resp.json().get('results', [])
    
    if transactions:
        acct_types = set(t.get('accountType', 'unknown') for t in transactions)
        detail = (f'WASH SALE RISK: {ticker} purchased within 30-day window '
                  f'in account types: {", ".join(acct_types)}. '
                  f'{len(transactions)} conflicting transaction(s) found.')
        return True, detail
    
    # Also check substitute security (substantially identical concern)
    sub = SUBSTITUTE_MAP.get(ticker, {}).get('sub')
    if sub:
        resp2 = requests.get(
            f'{api_base}/portfolio/transactions',
            headers=headers,
            params={
                'householdId': household_id,
                'ticker': sub,
                'transactionType': 'buy',
                'startDate': window_start,
                'endDate': window_end,
                'includeAllAccountTypes': True
            }
        )
        resp2.raise_for_status()
        sub_transactions = resp2.json().get('results', [])
        if sub_transactions:
            detail = (f'WASH SALE RISK: Substitute {sub} purchased within '
                      f'30-day window. Cannot use this substitute.')
            return True, detail
    
    return False, 'No wash-sale risk detected.'


def calculate_priority_score(loss_amount: float, tax_profile: TaxProfile,
                              loss_type: str, wash_sale_risk: bool) -> tuple:
    """Calculate TLH priority score (0-100) and estimated tax benefit.
    
    Scoring factors:
    1. Tax benefit magnitude (40% weight) - higher marginal rate = higher benefit
    2. Loss amount (30% weight) - larger losses = higher priority
    3. Loss type (15% weight) - short-term losses offset higher-taxed ST gains
    4. Risk/complexity (15% weight) - penalize wash-sale risk
    """
    # Calculate estimated tax benefit
    combined_rate = tax_profile.marginal_rate + tax_profile.state_rate
    if loss_type == 'short_term':
        applicable_rate = tax_profile.stcg_rate + tax_profile.state_rate
    else:
        applicable_rate = tax_profile.ltcg_rate + tax_profile.state_rate
    
    tax_benefit = abs(loss_amount) * applicable_rate
    
    # If client has existing carryforward, reduce priority slightly
    # (they already have losses to use)
    carryforward_penalty = min(tax_profile.capital_loss_carryforward / 50000, 0.2)
    
    # Score component 1: Tax benefit magnitude (0-40 points)
    # Scale: $0 = 0 pts, $5000+ = 40 pts
    benefit_score = min(tax_benefit / 5000, 1.0) * 40
    
    # Score component 2: Loss amount (0-30 points)
    # Scale: $0 = 0 pts, $25000+ = 30 pts
    loss_score = min(abs(loss_amount) / 25000, 1.0) * 30
    
    # Score component 3: Loss type (0-15 points)
    type_score = 15 if loss_type == 'short_term' else 8
    
    # Score component 4: Risk penalty (0-15 points)
    risk_score = 0 if wash_sale_risk else 15
    
    raw_score = benefit_score + loss_score + type_score + risk_score
    adjusted_score = raw_score * (1 - carryforward_penalty)
    
    return round(adjusted_score, 1), round(tax_benefit, 2)


def scan_for_opportunities(holdings: list, tax_profiles: dict,
                           api_base: str, api_key: str) -> List[TLHOpportunity]:
    """Main scanning engine: evaluate all holdings for TLH opportunities."""
    opportunities = []
    
    for holding in holdings:
        ticker = holding.get('ticker', '')
        unrealized_gl = holding.get('unrealizedGainLoss', 0)
        
        # Skip if no loss or loss below threshold
        if unrealized_gl >= -MIN_LOSS_THRESHOLD:
            continue
        
        # Skip if no substitute available
        if ticker not in SUBSTITUTE_MAP:
            logger.debug(f'No substitute mapped for {ticker}, skipping')
            continue
        
        # Skip if holding period too short
        purchase_date = datetime.strptime(
            holding.get('purchaseDate', '2020-01-01'), '%Y-%m-%d')
        holding_days = (datetime.now() - purchase_date).days
        if holding_days < MIN_HOLDING_PERIOD_DAYS:
            continue
        
        # Determine loss type
        loss_type = 'short_term' if holding_days <= 365 else 'long_term'
        
        # Check wash-sale risk
        household_id = holding.get('householdId', '')
        wash_risk, wash_detail = check_wash_sale_risk(
            household_id, ticker, api_base, api_key)
        
        # Get tax profile (use defaults if not found)
        tax_profile = tax_profiles.get(household_id, TaxProfile(
            household_id=household_id,
            client_name=holding.get('clientName', 'Unknown'),
            marginal_rate=0.24, ltcg_rate=0.15, stcg_rate=0.24,
            capital_loss_carryforward=0, amt_flag=False, state_rate=0.05
        ))
        
        # Calculate priority score
        priority_score, tax_benefit = calculate_priority_score(
            unrealized_gl, tax_profile, loss_type, wash_risk)
        
        sub = SUBSTITUTE_MAP[ticker]
        opportunities.append(TLHOpportunity(
            household_id=household_id,
            client_name=tax_profile.client_name,
            account_id=holding.get('accountId', ''),
            account_name=holding.get('accountName', ''),
            security_ticker=ticker,
            security_name=holding.get('securityName', ''),
            quantity=holding.get('quantity', 0),
            cost_basis=holding.get('costBasis', 0),
            current_value=holding.get('marketValue', 0),
            unrealized_loss=unrealized_gl,
            loss_type=loss_type,
            holding_period_days=holding_days,
            substitute_ticker=sub['sub'],
            substitute_name=sub['name'],
            wash_sale_risk=wash_risk,
            wash_sale_detail=wash_detail if wash_risk else None,
            tax_benefit_estimate=tax_benefit,
            priority_score=priority_score
        ))
    
    # Sort by priority score descending
    opportunities.sort(key=lambda x: x.priority_score, reverse=True)
    return opportunities


def generate_html_report(opportunities: List[TLHOpportunity],
                         report_type: str = 'daily') -> str:
    """Generate HTML email report of TLH opportunities."""
    total_losses = sum(abs(o.unrealized_loss) for o in opportunities)
    total_benefit = sum(o.tax_benefit_estimate for o in opportunities)
    wash_sale_count = sum(1 for o in opportunities if o.wash_sale_risk)
    
    html = f'''
    <html><body style="font-family: Arial, sans-serif; max-width: 900px;">
    <h2>Tax-Loss Harvesting Opportunity Report - {report_type.title()}</h2>
    <p>Generated: {datetime.now().strftime('%B %d, %Y at %I:%M %p ET')}</p>
    
    <div style="background: #f0f7ff; padding: 15px; border-radius: 8px; margin: 15px 0;">
        <h3 style="margin-top:0;">Summary</h3>
        <table>
            <tr><td><strong>Total Opportunities:</strong></td><td>{len(opportunities)}</td></tr>
            <tr><td><strong>Total Harvestable Losses:</strong></td><td>${total_losses:,.2f}</td></tr>
            <tr><td><strong>Estimated Tax Benefit:</strong></td><td>${total_benefit:,.2f}</td></tr>
            <tr><td><strong>Wash-Sale Flags:</strong></td>
                <td style="color: {'red' if wash_sale_count > 0 else 'green'};">
                    {wash_sale_count} {'⚠️' if wash_sale_count > 0 else '✅'}</td></tr>
        </table>
    </div>
    
    <table style="border-collapse: collapse; width: 100%; font-size: 12px;">
    <tr style="background: #2c3e50; color: white;">
        <th style="padding:8px;">Priority</th>
        <th style="padding:8px;">Client</th>
        <th style="padding:8px;">Account</th>
        <th style="padding:8px;">Security</th>
        <th style="padding:8px;">Loss</th>
        <th style="padding:8px;">Type</th>
        <th style="padding:8px;">Tax Benefit</th>
        <th style="padding:8px;">Substitute</th>
        <th style="padding:8px;">Wash Sale</th>
    </tr>
    '''
    
    for i, opp in enumerate(opportunities[:50]):  # Top 50
        bg = '#fff3f3' if opp.wash_sale_risk else ('#f9f9f9' if i % 2 else '#ffffff')
        ws_icon = '⚠️ RISK' if opp.wash_sale_risk else '✅ Clear'
        html += f'''
        <tr style="background: {bg};">
            <td style="padding:6px; text-align:center; font-weight:bold;">{opp.priority_score}</td>
            <td style="padding:6px;">{opp.client_name}</td>
            <td style="padding:6px;">{opp.account_name}</td>
            <td style="padding:6px;"><strong>{opp.security_ticker}</strong></td>
            <td style="padding:6px; color:red;">${abs(opp.unrealized_loss):,.2f}</td>
            <td style="padding:6px;">{opp.loss_type.replace('_', ' ').title()}</td>
            <td style="padding:6px; color:green;">${opp.tax_benefit_estimate:,.2f}</td>
            <td style="padding:6px;">{opp.substitute_ticker}</td>
            <td style="padding:6px;">{ws_icon}</td>
        </tr>
        '''
    
    html += '''</table>
    <p style="font-size:11px; color:#666; margin-top:20px;">
    This report is generated by automated TLH scanning software for advisor review only.
    All harvesting decisions require advisor approval. Wash-sale flagged items must NOT 
    be executed without manual review. This is not investment advice.
    </p>
    </body></html>'''
    
    return html


def send_alert(html_content: str, subject: str, recipients: list,
               include_cco: bool = False):
    """Send email alert via SMTP."""
    msg = MIMEMultipart('alternative')
    msg['Subject'] = subject
    msg['From'] = SMTP_USER
    all_recipients = recipients.copy()
    if include_cco:
        all_recipients.append(CCO_EMAIL)
    msg['To'] = ', '.join(all_recipients)
    msg.attach(MIMEHTML(html_content, 'html'))
    
    with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
        server.starttls()
        server.login(SMTP_USER, SMTP_PASS)
        server.sendmail(SMTP_USER, all_recipients, msg.as_string())
    logger.info(f'Alert sent to {len(all_recipients)} recipients: {subject}')


def check_market_correction() -> bool:
    """Check if S&P 500 has declined more than threshold in trailing 30 days.
    Uses a simple API call to check market conditions."""
    try:
        # Using Yahoo Finance or similar free endpoint
        resp = requests.get(
            'https://query1.finance.yahoo.com/v8/finance/chart/SPY',
            params={'range': '1mo', 'interval': '1d'},
            headers={'User-Agent': 'TLH-Scanner/1.0'}
        )
        data = resp.json()
        prices = data['chart']['result'][0]['indicators']['quote'][0]['close']
        prices = [p for p in prices if p is not None]
        if len(prices) >= 2:
            change = (prices[-1] - prices[0]) / prices[0]
            return change <= MARKET_CORRECTION_THRESHOLD
    except Exception as e:
        logger.warning(f'Market check failed: {e}')
    return False


def main():
    """Main execution: daily TLH scan and alerting."""
    logger.info('Starting TLH opportunity scan...')
    
    # Load tax profiles
    tax_profiles = load_tax_profiles()
    logger.info(f'Loaded {len(tax_profiles)} tax profiles')
    
    # Fetch holdings from Orion
    holdings = get_orion_holdings(ORION_API_BASE, ORION_API_KEY)
    logger.info(f'Fetched {len(holdings)} holdings from Orion')
    
    # Scan for opportunities
    opportunities = scan_for_opportunities(
        holdings, tax_profiles, ORION_API_BASE, ORION_API_KEY)
    logger.info(f'Found {len(opportunities)} TLH opportunities')
    
    # Determine report type and urgency
    today = datetime.now()
    is_q4_push = today.month >= 10 and today.month <= 12
    is_market_correction = check_market_correction()
    is_monday = today.weekday() == 0
    
    if is_market_correction:
        subject = '🚨 URGENT: Market Correction TLH Opportunities'
        report_type = 'market_correction'
        include_cco = True
    elif is_q4_push:
        subject = f'📊 Q4 Year-End TLH Opportunities - {today.strftime("%b %d")}'
        report_type = 'q4_push'
        include_cco = False
    elif is_monday:
        subject = f'📋 Weekly TLH Summary - Week of {today.strftime("%b %d")}'
        report_type = 'weekly'
        include_cco = True
    else:
        subject = f'📊 Daily TLH Scan - {today.strftime("%b %d")}'
        report_type = 'daily'
        include_cco = False
    
    if opportunities:
        html = generate_html_report(opportunities, report_type)
        send_alert(html, subject, ALERT_RECIPIENTS, include_cco)
    elif is_monday:  # Always send Monday summary even if no opportunities
        html = generate_html_report([], report_type)
        send_alert(html, subject + ' (No Opportunities Found)',
                   ALERT_RECIPIENTS, include_cco)
    
    logger.info('TLH scan complete.')


if __name__ == '__main__':
    main()

# function.json. Schedule: 6:30 AM ET (10:30 UTC) Monday–Friday.

{
  "bindings": [
    {
      "name": "timer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "0 30 10 * * 1-5"
    }
  ]
}

requests>=2.31.0
python-dateutil>=2.8.2

# Configuration file for the 4 seasonal TLH campaigns. Deploy as Redtail
# workflows + calendar-triggered automation.

campaigns:
  q1_january_effect:
    name: "Q1 January Effect TLH Sweep"
    trigger_date: "January 2"
    end_date: "January 31"
    description: >
      Capitalize on January selling pressure and year-start volatility.
      Many securities experience price dips in early January as investors
      sell winners from prior year for cash flow or rebalancing.
    scan_frequency: "daily"
    actions:
      - action: "enhanced_scan"
        description: "Run TLH Priority Scorer daily with lowered thresholds"
        min_loss_override: 50  # Lower threshold during high-opportunity window
      - action: "redtail_workflow"
        workflow_name: "Q1 January TLH Review"
        steps:
          - assignee: "operations"
            task: "Generate TLH opportunity report from Orion Custom Indexing"
            due: "January 3"
          - assignee: "lead_advisor"
            task: "Review top 20 priority opportunities and approve/reject"
            due: "January 5"
          - assignee: "operations"
            task: "Execute approved TLH trades in Orion"
            due: "January 6"
          - assignee: "operations"
            task: "Document all trades in Redtail using TLH Trade Documentation template"
            due: "January 7"
          - assignee: "cco"
            task: "Review January TLH batch for compliance"
            due: "January 10"
      - action: "client_communication"
        template: "q1_tlh_proactive"
        subject: "Starting the Year Smart: Tax-Efficient Portfolio Review"
        send_to: "households_with_executed_tlh"
        timing: "After trades execute"

  spring_tax_return_review:
    name: "Spring Tax Return TLH Intelligence Refresh"
    trigger_date: "April 20"
    end_date: "May 31"
    description: >
      Annual refresh of client tax intelligence using newly filed tax returns.
      Update marginal rates, capital loss carryforwards, and AMT exposure
      in Holistiplan. Recalculate TLH priority scores with fresh data.
    scan_frequency: "weekly"
    actions:
      - action: "holistiplan_refresh"
        description: "Upload all new client 1040s to Holistiplan"
        steps:
          - "Collect digital copies of all client tax returns from filing"
          - "Batch upload to Holistiplan (up to 50 at a time)"
          - "Export updated tax intelligence CSV"
          - "Update Redtail custom fields: marginal rate, LTCG rate, loss carryforward"
          - "Update tax_profiles.csv for TLH Priority Scoring Engine"
      - action: "redtail_workflow"
        workflow_name: "Spring Tax Intelligence Update"
        steps:
          - assignee: "operations"
            task: "Collect all 2024 tax returns from clients/CPAs"
            due: "April 25"
          - assignee: "operations"
            task: "Upload returns to Holistiplan and export data"
            due: "May 1"
          - assignee: "operations"
            task: "Update Redtail custom fields for all households"
            due: "May 5"
          - assignee: "lead_advisor"
            task: "Review clients with significant rate changes or new carryforwards"
            due: "May 10"
          - assignee: "lead_advisor"
            task: "Schedule tax planning meetings for high-priority households"
            due: "May 15"

  q3_midyear_check:
    name: "Q3 Mid-Year TLH and Rebalance Check"
    trigger_date: "July 1"
    end_date: "July 31"
    description: >
      Mid-year review combining rebalancing with TLH opportunity identification.
      Summer is often overlooked for TLH but can yield significant opportunities
      especially in volatile sectors or individual positions.
    scan_frequency: "weekly"
    actions:
      - action: "enhanced_scan"
        description: "Run combined rebalance + TLH analysis"
      - action: "redtail_workflow"
        workflow_name: "Q3 Mid-Year TLH Review"
        steps:
          - assignee: "operations"
            task: "Generate mid-year TLH report and YTD harvested losses summary"
            due: "July 5"
          - assignee: "lead_advisor"
            task: "Review mid-year opportunities; prioritize accounts under-harvested YTD"
            due: "July 10"
          - assignee: "lead_advisor"
            task: "Review BlackRock Tax Evaluator for early capital gains distribution estimates"
            due: "July 15"
          - assignee: "operations"
            task: "Execute approved mid-year TLH trades"
            due: "July 18"
          - assignee: "cco"
            task: "Mid-year compliance review of all TLH activity YTD"
            due: "July 25"

  q4_yearend_push:
    name: "Q4 Year-End TLH Push"
    trigger_date: "October 1"
    end_date: "December 20"
    description: >
      The most critical TLH period. Enhanced daily scanning, proactive client
      outreach, capital gains distribution monitoring, and final harvesting
      before year-end. Last trade date typically December 29-30.
    scan_frequency: "daily"
    priority: "highest"
    actions:
      - action: "blackrock_screen"
        description: "Screen all client mutual fund holdings through BlackRock Tax Evaluator"
        timing: "October 1-15"
        steps:
          - "Export all mutual fund/ETF holdings from Orion"
          - "Screen each fund through BlackRock Tax Evaluator"
          - "Flag funds with estimated distributions >3% of NAV"
          - "Create proactive TLH plan for affected accounts"
      - action: "enhanced_scan"
        description: "Daily TLH scanning with Q4 urgency alerts"
        min_loss_override: 50
        alert_escalation: true
      - action: "redtail_workflow"
        workflow_name: "Q4 Year-End TLH Campaign"
        steps:
          - assignee: "operations"
            task: "BlackRock Tax Evaluator screening of all fund holdings"
            due: "October 10"
          - assignee: "lead_advisor"
            task: "Review capital gains distribution exposure report"
            due: "October 15"
          - assignee: "advisor_team"
            task: "Client outreach: year-end tax planning discussions"
            due: "October 31"
          - assignee: "operations"
            task: "Execute first wave of Q4 TLH trades"
            due: "November 15"
          - assignee: "lead_advisor"
            task: "Final review: last chance TLH opportunities"
            due: "December 10"
          - assignee: "operations"
            task: "Execute final TLH trades (last trading day minus 3)"
            due: "December 20"
          - assignee: "operations"
            task: "Generate annual TLH performance report by household"
            due: "December 31"
          - assignee: "cco"
            task: "Q4 and annual TLH compliance review"
            due: "January 10 (next year)"
      - action: "client_communication"
        template: "q4_yearend_results"
        subject: "Your 2024 Tax-Smart Portfolio Summary"
        send_to: "all_tlh_eligible_households"
        timing: "January 5-10 (next year)"

# Email Templates
templates:
  q1_tlh_proactive:
    subject: "Starting {year} Smart: Tax-Efficient Portfolio Review"
    body: >
      Dear {client_name},
      
      As part of our ongoing commitment to maximizing your after-tax returns,
      we've completed an early-year review of your portfolio and identified
      opportunities to improve your tax position.
      
      We recently executed tax-loss harvesting trades in your account(s) that
      are expected to generate approximately ${estimated_benefit} in tax
      savings for {year}. These trades maintain your target allocation while
      capturing available tax losses.
      
      We'll continue monitoring your portfolio throughout the year for
      additional opportunities. Please don't hesitate to reach out if you
      have questions.
      
      Best regards,
      {advisor_name}

  q4_yearend_results:
    subject: "Your {year} Tax-Smart Portfolio Summary"
    body: >
      Dear {client_name},
      
      As we close out {year}, I'm pleased to share a summary of the
      tax-efficiency efforts we've undertaken on your behalf this year:
      
      • Total tax losses harvested: ${total_losses_harvested}
      • Estimated tax benefit: ${estimated_tax_benefit}
      • Number of harvesting events: {harvest_count}
      • Portfolio tracking error: maintained within target
      
      These losses can offset capital gains realized this year and, if
      unused, up to $3,000 of ordinary income, with the remainder carrying
      forward to future years.
      
      I'd like to schedule a brief call to discuss your year-end tax
      position and planning for {next_year}. Please let me know a
      convenient time.
      
      Best regards,
      {advisor_name}

import requests
import json
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Tuple, Optional
from enum import Enum

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('wash_sale_guardian')


class WashSaleVerdict(Enum):
    PASS = 'PASS'
    FAIL_SAME_SECURITY = 'FAIL_SAME_SECURITY'
    FAIL_SUBSTANTIALLY_IDENTICAL = 'FAIL_SUBSTANTIALLY_IDENTICAL'
    FAIL_SPOUSE_ACCOUNT = 'FAIL_SPOUSE_ACCOUNT'
    FAIL_IRA_PURCHASE = 'FAIL_IRA_PURCHASE'
    WARNING_PENDING_ORDER = 'WARNING_PENDING_ORDER'
    ERROR_DATA_UNAVAILABLE = 'ERROR_DATA_UNAVAILABLE'


@dataclass
class WashSaleCheckResult:
    verdict: WashSaleVerdict
    ticker_checked: str
    household_id: str
    conflicting_transactions: list
    conflicting_accounts: list
    explanation: str
    check_timestamp: str
    window_start: str
    window_end: str
    recommendation: str


# Substantially identical security groups
# Securities within the same group are considered substantially identical
# for wash-sale purposes. This is a CONSERVATIVE interpretation.
SUBSTANTIALLY_IDENTICAL_GROUPS = {
    'sp500': ['SPY', 'IVV', 'VOO', 'SPLG', 'BBUS'],
    'total_us': ['VTI', 'ITOT', 'SCHB', 'SPTM'],
    'nasdaq100': ['QQQ', 'QQQM'],
    'intl_dev': ['VXUS', 'IXUS', 'VEA', 'EFA', 'SPDW', 'IEFA'],
    'intl_em': ['VWO', 'IEMG', 'EEM', 'SCHE'],
    'total_bond': ['BND', 'AGG', 'SCHZ', 'FBND'],
    'us_reit': ['VNQ', 'SCHH', 'IYR', 'XLRE'],
    'us_dividend': ['VIG', 'DGRO', 'SCHD', 'DVY'],
    'us_growth': ['VUG', 'IWF', 'SCHG', 'SPYG'],
    'us_value': ['VTV', 'IWD', 'SCHV', 'SPYV'],
    'us_smallcap': ['VB', 'IJR', 'SCHA', 'IWM'],
    'tips': ['TIP', 'SCHP', 'VTIP'],
    'short_term_bond': ['BSV', 'SHY', 'SCHO', 'VGSH'],
}


def get_identical_group(ticker: str) -> Optional[set]:
    """Find all substantially identical securities for a given ticker."""
    for group_name, members in SUBSTANTIALLY_IDENTICAL_GROUPS.items():
        if ticker.upper() in [m.upper() for m in members]:
            return set(m.upper() for m in members)
    return None


def check_wash_sale(
    ticker: str,
    household_id: str,
    proposed_sale_date: datetime,
    orion_api_base: str,
    orion_api_key: str
) -> WashSaleCheckResult:
    """Comprehensive wash-sale rule check.
    
    Checks:
    1. Same security purchases within 30 days before/after in ALL accounts
    2. Substantially identical security purchases in ALL accounts
    3. Spouse/partner accounts within the household
    4. Tax-deferred accounts (IRA, Roth IRA, 401(k))
    5. Pending/open orders for the same or substantially identical securities
    
    Returns WashSaleCheckResult with verdict and full audit trail.
    """
    headers = {
        'Authorization': f'Bearer {orion_api_key}',
        'Content-Type': 'application/json'
    }
    
    check_timestamp = datetime.now().isoformat()
    window_start = (proposed_sale_date - timedelta(days=30)).strftime('%Y-%m-%d')
    window_end = (proposed_sale_date + timedelta(days=30)).strftime('%Y-%m-%d')
    
    # Build set of tickers to check (same + substantially identical)
    tickers_to_check = {ticker.upper()}
    identical_group = get_identical_group(ticker)
    if identical_group:
        tickers_to_check = tickers_to_check.union(identical_group)
    
    all_conflicts = []
    conflict_accounts = []
    
    try:
        # Get all accounts in household (including spouse, IRAs, 401(k)s)
        acct_resp = requests.get(
            f'{orion_api_base}/household/{household_id}/accounts',
            headers=headers,
            params={'includeAllTypes': True}
        )
        acct_resp.raise_for_status()
        household_accounts = acct_resp.json().get('results', [])
        
        for account in household_accounts:
            account_id = account.get('accountId')
            account_name = account.get('accountName', 'Unknown')
            account_type = account.get('registrationType', 'Unknown')
            owner = account.get('ownerName', 'Unknown')
            
            for check_ticker in tickers_to_check:
                # Check completed transactions
                txn_resp = requests.get(
                    f'{orion_api_base}/account/{account_id}/transactions',
                    headers=headers,
                    params={
                        'ticker': check_ticker,
                        'transactionType': 'buy',
                        'startDate': window_start,
                        'endDate': window_end
                    }
                )
                txn_resp.raise_for_status()
                transactions = txn_resp.json().get('results', [])
                
                for txn in transactions:
                    conflict = {
                        'type': 'completed_transaction',
                        'ticker': check_ticker,
                        'account_id': account_id,
                        'account_name': account_name,
                        'account_type': account_type,
                        'owner': owner,
                        'transaction_date': txn.get('transactionDate'),
                        'quantity': txn.get('quantity'),
                        'amount': txn.get('amount'),
                        'is_same_security': check_ticker.upper() == ticker.upper(),
                        'is_tax_deferred': account_type in [
                            'IRA', 'Roth IRA', 'SEP IRA', 'SIMPLE IRA',
                            '401(k)', '403(b)', '457(b)'
                        ]
                    }
                    all_conflicts.append(conflict)
                    if account_id not in conflict_accounts:
                        conflict_accounts.append({
                            'account_id': account_id,
                            'account_name': account_name,
                            'account_type': account_type,
                            'owner': owner
                        })
                
                # Check pending/open orders
                order_resp = requests.get(
                    f'{orion_api_base}/account/{account_id}/orders',
                    headers=headers,
                    params={
                        'ticker': check_ticker,
                        'status': 'open,pending',
                        'side': 'buy'
                    }
                )
                if order_resp.status_code == 200:
                    open_orders = order_resp.json().get('results', [])
                    for order in open_orders:
                        conflict = {
                            'type': 'pending_order',
                            'ticker': check_ticker,
                            'account_id': account_id,
                            'account_name': account_name,
                            'account_type': account_type,
                            'owner': owner,
                            'order_date': order.get('orderDate'),
                            'quantity': order.get('quantity'),
                            'is_same_security': check_ticker.upper() == ticker.upper()
                        }
                        all_conflicts.append(conflict)
    
    except requests.RequestException as e:
        logger.error(f'API error during wash-sale check: {e}')
        return WashSaleCheckResult(
            verdict=WashSaleVerdict.ERROR_DATA_UNAVAILABLE,
            ticker_checked=ticker,
            household_id=household_id,
            conflicting_transactions=[],
            conflicting_accounts=[],
            explanation=f'Unable to complete wash-sale check due to API error: {str(e)}',
            check_timestamp=check_timestamp,
            window_start=window_start,
            window_end=window_end,
            recommendation='DO NOT EXECUTE. Manual review required. API connectivity issue.'
        )
    
    # Determine verdict
    if not all_conflicts:
        return WashSaleCheckResult(
            verdict=WashSaleVerdict.PASS,
            ticker_checked=ticker,
            household_id=household_id,
            conflicting_transactions=[],
            conflicting_accounts=[],
            explanation=(
                f'No purchases of {ticker} or substantially identical securities '
                f'found in any household account within the 61-day wash-sale window '
                f'({window_start} to {window_end}). '
                f'Checked {len(tickers_to_check)} tickers across '
                f'{len(household_accounts)} accounts.'
            ),
            check_timestamp=check_timestamp,
            window_start=window_start,
            window_end=window_end,
            recommendation='SAFE TO EXECUTE. No wash-sale risk detected.'
        )
    
    # Categorize the failure
    has_ira_conflict = any(c.get('is_tax_deferred') for c in all_conflicts)
    has_same_security = any(c.get('is_same_security') for c in all_conflicts)
    has_pending = any(c.get('type') == 'pending_order' for c in all_conflicts)
    
    if has_pending and not any(c.get('type') == 'completed_transaction' for c in all_conflicts):
        verdict = WashSaleVerdict.WARNING_PENDING_ORDER
        recommendation = (
            'CANCEL PENDING BUY ORDERS before executing TLH sale, or '
            'wait until pending orders are resolved. Manual review required.'
        )
    elif has_ira_conflict:
        verdict = WashSaleVerdict.FAIL_IRA_PURCHASE
        recommendation = (
            'DO NOT EXECUTE. Wash-sale rule applies even when the replacement '
            'purchase is in a tax-deferred account. The loss will be permanently '
            'disallowed (cannot be added to IRA cost basis). Wait 31 days after '
            'the most recent purchase date.'
        )
    elif has_same_security:
        verdict = WashSaleVerdict.FAIL_SAME_SECURITY
        recommendation = (
            'DO NOT EXECUTE. Same security purchased within wash-sale window. '
            f'Wait until after {window_end} or use a different substitute security '
            'that is not in the substantially identical group.'
        )
    else:
        verdict = WashSaleVerdict.FAIL_SUBSTANTIALLY_IDENTICAL
        recommendation = (
            'DO NOT EXECUTE. Substantially identical security purchased within '
            'wash-sale window. Consider using a substitute from a different '
            'asset class or strategy, or wait until after the wash-sale window closes.'
        )
    
    explanation_parts = [f'Found {len(all_conflicts)} conflicting transaction(s):']
    for c in all_conflicts:
        explanation_parts.append(
            f"  - {c['type']}: {c['ticker']} in {c['account_name']} "
            f"({c['account_type']}, owner: {c['owner']}) "
            f"on {c.get('transaction_date', c.get('order_date', 'N/A'))}"
        )
    
    return WashSaleCheckResult(
        verdict=verdict,
        ticker_checked=ticker,
        household_id=household_id,
        conflicting_transactions=all_conflicts,
        conflicting_accounts=conflict_accounts,
        explanation='\n'.join(explanation_parts),
        check_timestamp=check_timestamp,
        window_start=window_start,
        window_end=window_end,
        recommendation=recommendation
    )


def validate_tlh_batch(
    proposed_trades: list,
    orion_api_base: str,
    orion_api_key: str
) -> dict:
    """Validate an entire batch of proposed TLH trades.
    Returns a summary with per-trade verdicts.
    
    proposed_trades format:
    [
        {'ticker': 'SPY', 'household_id': 'HH001', 'sale_date': '2025-01-15'},
        ...
    ]
    """
    results = {
        'batch_timestamp': datetime.now().isoformat(),
        'total_trades': len(proposed_trades),
        'passed': 0,
        'failed': 0,
        'warnings': 0,
        'errors': 0,
        'trade_results': []
    }
    
    for trade in proposed_trades:
        result = check_wash_sale(
            ticker=trade['ticker'],
            household_id=trade['household_id'],
            proposed_sale_date=datetime.strptime(trade['sale_date'], '%Y-%m-%d'),
            orion_api_base=orion_api_base,
            orion_api_key=orion_api_key
        )
        
        trade_result = {
            'ticker': trade['ticker'],
            'household_id': trade['household_id'],
            'verdict': result.verdict.value,
            'explanation': result.explanation,
            'recommendation': result.recommendation
        }
        results['trade_results'].append(trade_result)
        
        if result.verdict == WashSaleVerdict.PASS:
            results['passed'] += 1
        elif result.verdict.value.startswith('WARNING'):
            results['warnings'] += 1
        elif result.verdict.value.startswith('ERROR'):
            results['errors'] += 1
        else:
            results['failed'] += 1
    
    # Safety gate: if ANY trade fails, flag entire batch for manual review
    results['batch_verdict'] = 'APPROVED' if results['failed'] == 0 and results['errors'] == 0 else 'REQUIRES_MANUAL_REVIEW'
    
    logger.info(
        f"Batch validation complete: {results['passed']} passed, "
        f"{results['failed']} failed, {results['warnings']} warnings, "
        f"{results['errors']} errors. Batch verdict: {results['batch_verdict']}"
    )
    
    return results

### Client Information
- **Household Name:** {household_name}
- **Primary Contact:** {primary_contact_name}
- **Advisor:** {advisor_name}
- **Reporting Period:** {period_start} to {period_end}
- **Household AUM:** ${household_aum}
- **Number of Taxable Accounts:** {taxable_account_count}
- **Marginal Federal Tax Rate:** {marginal_rate}%
- **State Tax Rate:** {state_rate}%
- **Long-Term Capital Gains Rate:** {ltcg_rate}%

### TLH Activity Summary
- **Total Harvesting Events:** {harvest_count}
- **Total Short-Term Losses Harvested:** ${st_losses}
- **Total Long-Term Losses Harvested:** ${lt_losses}
- **Total Losses Harvested:** ${total_losses}
- **Estimated Federal Tax Savings:** ${federal_tax_savings}
- **Estimated State Tax Savings:** ${state_tax_savings}
- **Total Estimated Tax Savings:** ${total_tax_savings}
- **Capital Loss Carryforward (if applicable):** ${carryforward_amount}
- **Wash-Sale Violations:** {wash_sale_violations} (should be 0)

### Harvesting Events Detail
{harvesting_events_table}
<!-- JSON array of: date, security_sold, loss_amount, loss_type, substitute_purchased -->

### Portfolio Impact
- **Portfolio Tracking Error vs. Benchmark:** {tracking_error}%
- **Portfolio Total Return (pre-tax):** {total_return}%
- **Benchmark Return:** {benchmark_return}%
- **Estimated After-Tax Alpha from TLH:** {after_tax_alpha} bps

# loads prompt template, injects client data variables, and calls the OpenAI
# Chat Completions API

import openai
import json

def generate_tlh_report(client_data: dict, api_key: str) -> str:
    """Generate client-facing TLH report using LLM."""
    client = openai.OpenAI(api_key=api_key)
    
    # Load prompt template and fill in variables
    with open('tlh_report_prompt.md', 'r') as f:
        prompt_template = f.read()
    
    # Replace template variables
    prompt = prompt_template
    for key, value in client_data.items():
        prompt = prompt.replace(f'{{{key}}}', str(value))
    
    response = client.chat.completions.create(
        model='gpt-5.4',
        messages=[
            {'role': 'system', 'content': prompt.split('## User Prompt Template')[0]},
            {'role': 'user', 'content': prompt.split('## User Prompt Template')[1]}
        ],
        temperature=0.3,
        max_tokens=3000
    )
    
    return response.choices[0].message.content