"""Fixed-rule AMZN PMCC on observed LSEG daily option quotes, calendar 2024.

One pair, no leverage. Yesterday chooses the contract; today's closing bid/ask
sets a simulated fill. No interpolation, last-trade fills or invented prices.
Raw LSEG/WRDS inputs remain local. This script publishes derived book results.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import platform
import shutil
import zipfile
from dataclasses import asdict, dataclass, replace
from pathlib import Path

import numpy as np
import pandas as pd
import psycopg

ROOT = Path(__file__).resolve().parents[1]
START, END = '2024-01-02', '2024-12-31'
DATA = ROOT.parent.parent / '06_Local_Data' / 'LSEG_PMCC'


@dataclass(frozen=True)
class Policy:
    starting_cash: float = 50_000
    long_delta: float = .80
    short_delta: float = .25
    long_target_days: int = 450
    short_target_days: int = 35
    long_roll_days: int = 180
    short_close_days: int = 7
    short_defense_delta: float = .60
    execution_fraction: float = 1.0  # 0=mid, 1=full observed half-spread.
    option_fee: float = .65
    stock_half_spread: float = .01
    minimum_oi: int = 100


def read_quotes(directory=DATA):
    frames, manifests = [], []
    for path in sorted(Path(directory).glob('*.json')):
        manifest = json.loads(path.read_text())
        if manifest.get('request', {}).get('start') != '2023-12-26':
            continue  # Exclude the small connection probe.
        csv = path.with_suffix('.csv')
        if hashlib.sha256(csv.read_bytes()).hexdigest() != manifest['sha256']:
            raise ValueError(f'Input fingerprint changed: {csv.name}')
        frames.append(pd.read_csv(csv))
        manifests.append({'file': csv.name, 'sha256': manifest['sha256'], 'request': manifest['request']})
    if not frames:
        raise ValueError('No observed option extract; run scripts/fetch_pmcc_lseg.py first')
    tidy = pd.concat(frames, ignore_index=True).drop_duplicates(['date', 'ric', 'field'])
    quotes = tidy.pivot(index=['date', 'ric', 'expiry', 'strike'], columns='field', values='value').reset_index()
    quotes.columns.name = None
    quotes['date'] = pd.to_datetime(quotes.date)
    quotes['expiry'] = pd.to_datetime(quotes.expiry)
    quotes['mid'] = (quotes.BID + quotes.ASK) / 2
    quotes['valid'] = (quotes.BID >= 0) & (quotes.ASK > 0) & (quotes.ASK >= quotes.BID)
    quotes['relativeSpread'] = (quotes.ASK - quotes.BID) / quotes['mid']
    quotes['dte'] = (quotes.expiry - quotes.date).dt.days
    return quotes.sort_values(['date', 'ric']).reset_index(drop=True), manifests


def read_stock(dsn):
    with psycopg.connect(dsn) as conn:
        rows = conn.execute("""SELECT trade_date, close, cumulative_adjustment_factor, cumulative_return_factor
            FROM optionmetrics_security_prices WHERE ticker='AMZN'
            AND source_id='duke_wrds_optionmetrics_conditional'
            AND trade_date BETWEEN '2023-12-26' AND '2024-12-31' ORDER BY trade_date""").fetchall()
    return pd.DataFrame(rows, columns=['trade_date', 'close', 'cumulative_adjustment_factor', 'cumulative_return_factor'])


def replay(stock, quotes, kind='pmcc', p=Policy(), start=START, end=END):
    stock = stock.copy().sort_values('trade_date').reset_index(drop=True)
    stock['trade_date'] = pd.to_datetime(stock.trade_date)
    if stock.trade_date.duplicated().any() or (stock.close <= 0).any() or not np.isfinite(stock.close.astype(float)).all():
        raise ValueError('Invalid stock sessions')
    for col in ('cumulative_adjustment_factor', 'cumulative_return_factor'):
        if col in stock and stock[col].nunique() != 1:
            raise ValueError('Split or distribution: explicit adjustment required')
    stock['signal_date'] = stock.trade_date.shift(1)
    stock['prior_close'] = stock.close.shift(1)
    stock = stock[stock.trade_date.between(start, end)]
    if stock.empty or stock.signal_date.isna().any():
        raise ValueError('A completed prior stock session is required')
    days = {day: group.set_index('ric', drop=False) for day, group in quotes.groupby('date')}
    if quotes.duplicated(['date', 'ric']).any():
        raise ValueError('Duplicate option observation')
    cash = p.starting_cash
    long = short = None
    stock_qty = 0
    flows = {'long': 0., 'short': 0., 'stock': 0.}
    fills, ledger, decisions, outcomes = [], [], [], []
    fees = friction = 0.
    one_sided_fills = 0
    first_debit = None
    long_rolls = 0
    peak = p.starting_cash
    short_credit = 0.

    def quote(day, ric, require_both=True):
        table = days.get(day)
        if table is None or ric not in table.index or (require_both and not bool(table.loc[ric, 'valid'])):
            return None
        return table.loc[ric]

    def choose(day, execution_day, spot, leg, minimum=0):
        table = days.get(day)
        if table is None:
            return None
        # The as-of snapshot decides; no future quote availability enters ranking.
        dte = (table.expiry - execution_day).dt.days
        mask = table.valid & table.DELTA.between(0, 1) & (table.OPINT_1 >= p.minimum_oi)
        if leg == 'long':
            mask &= dte.between(365, 550) & (table.strike < spot) & (table.relativeSpread <= .10)
            target_days, target_delta = p.long_target_days, p.long_delta
        else:
            mask &= dte.between(28, 62) & (table.strike > max(spot, minimum)) & (table.BID > 0) & (table.relativeSpread <= .35)
            target_days, target_delta = p.short_target_days, p.short_delta
        candidate = table[mask].copy()
        if candidate.empty:
            return None
        candidate['expiryGap'] = abs(dte[mask] - target_days)
        candidate['deltaGap'] = abs(candidate.DELTA - target_delta)
        # First expiry nearest target, then delta nearest target within that expiry.
        selected = candidate.sort_values(['expiryGap', 'expiry', 'deltaGap', 'strike']).iloc[0]
        return {'id': selected.ric, 'strike': float(selected.strike), 'expiry': selected.expiry}

    def transact(day, signal, leg, side, contract, spot, reason):
        nonlocal cash, fees, friction, one_sided_fills
        qty, multiplier = (100, 1) if leg == 'stock' else (1, 100)
        if contract:
            q = quote(day, contract['id'], require_both=False)
            execution_side = 'ASK' if side == 'BUY' else 'BID'
            if q is None or not np.isfinite(q[execution_side]) or q[execution_side] < 0:
                raise ValueError(f'Missing execution quote: {day.date()} {contract["id"]}; no invented fill')
            if np.isfinite(q.BID) and np.isfinite(q.ASK) and (q.BID < 0 or q.ASK < q.BID):
                raise ValueError('Crossed or negative execution quote')
            if bool(q.valid):
                mid = float(q['mid']); half = float(q.ASK - q.BID) / 2
                concession = p.execution_fraction * half
                basis = 'Two-sided quote'
            else:
                # A buyback only needs the observed ask. Never turn a missing
                # bid into zero or claim a midpoint was observed. All scenarios
                # pay this ask; spread attribution is unavailable for this fill.
                mid, concession = float(q[execution_side]), 0.
                basis = f'{execution_side}-only; no midpoint assumption'
                one_sided_fills += 1
        else:
            mid, concession = spot, p.stock_half_spread
            basis = 'Stock close'
        price = mid + concession if side == 'BUY' else mid - concession
        fee = 0 if leg == 'stock' else p.option_fee
        flow = (1 if side == 'SELL' else -1) * qty * multiplier * price - fee
        if cash + flow < -1e-7:
            raise ValueError('Cash gate rejected borrowing')
        cash += flow; fees += fee; friction += concession * qty * multiplier
        flows[leg] += flow
        fills.append({'date': str(day.date()), 'time': 'EOD', 'signalDate': str(signal.date()), 'leg': leg,
                      'side': side, 'contract': contract['id'] if contract else 'AMZN · 100 shares',
                      'strike': contract['strike'] if contract else None,
                      'expiry': str(contract['expiry'].date()) if contract else None,
                      'quantity': qty, 'fair': round(mid, 5) if 'only' not in basis else None,
                      'quoteBasis': basis, 'fill': round(price, 5), 'fee': fee,
                      'friction': round(concession * qty * multiplier, 4), 'cashFlow': round(flow, 4),
                      'reason': reason, 'evidence': 'Observed LSEG EOD quote; simulated execution' if contract else 'Observed stock close + assumed 1-cent cost'})
        return flow

    for index, row in enumerate(stock.itertuples()):
        day, signal, spot = row.trade_date, row.signal_date, float(row.close)
        reasons = []
        renew = long is not None and (long['expiry'] - day).days <= p.long_roll_days
        if short:
            prior = quote(signal, short['id'], require_both=False)
            if prior is None or not np.isfinite(prior.DELTA):
                raise ValueError('Missing prior short quote/delta; decision cannot be reproduced')
            if renew or (short['expiry'] - day).days <= p.short_close_days or prior.DELTA >= p.short_defense_delta:
                reason = 'Long renewal' if renew else '7-DTE exit' if (short['expiry']-day).days <= p.short_close_days else 'Prior delta ≥ 0.60'
                flow = transact(day, signal, 'short', 'BUY', short, spot, reason)
                outcomes.append({'date': str(day.date()), 'contract': short['id'], 'pnl': round(short_credit + flow, 2)})
                short = None; reasons.append(reason)
        if renew:
            replacement = choose(signal, day, float(row.prior_close), 'long')
            if replacement is None or quote(day, replacement['id']) is None:
                raise ValueError('Long renewal unavailable; cannot pretend uninterrupted PMCC coverage')
            transact(day, signal, 'long', 'SELL', long, spot, 'Sell old long before renewal')
            transact(day, signal, 'long', 'BUY', replacement, spot, 'Renew long near 0.80 delta')
            long = replacement; long_rolls += 1; reasons.append('Renew long')
        if index == 0:
            if kind in ('pmcc', 'long'):
                selected = choose(signal, day, float(row.prior_close), 'long')
                if selected is None or quote(day, selected['id']) is None:
                    raise ValueError('Initial long unavailable on fixed start date')
                transact(day, signal, 'long', 'BUY', selected, spot, 'Initial 0.80-delta long')
                long = selected
            else:
                transact(day, signal, 'stock', 'BUY', None, spot, 'Initial 100 shares')
                stock_qty = 100
        if kind in ('pmcc', 'covered') and short is None and (index == 0 or day.weekday() == 0) and index < len(stock)-5:
            selected = choose(signal, day, float(row.prior_close), 'short', long['strike'] if long else 0)
            if selected is None or quote(day, selected['id']) is None:
                reasons.append('Skip: no eligible as-of contract / next-session quote')
            else:
                if long and selected['expiry'] >= long['expiry']:
                    raise ValueError('Short expiry must precede long expiry')
                short_credit = transact(day, signal, 'short', 'SELL', selected, spot, 'Prior snapshot: nearest 0.25 delta' if p.short_delta==.25 else f'Prior snapshot: nearest {p.short_delta:.2f} delta')
                short = selected; reasons.append('Write short')
        if index == 0:
            first_debit = p.starting_cash-cash
        if index == len(stock)-1:
            if short:
                flow = transact(day, signal, 'short', 'BUY', short, spot, 'Scheduled year-end liquidation')
                outcomes.append({'date': str(day.date()), 'contract': short['id'], 'pnl': round(short_credit+flow, 2)})
                short = None
            if long:
                transact(day, signal, 'long', 'SELL', long, spot, 'Scheduled year-end liquidation'); long = None
            if stock_qty:
                transact(day, signal, 'stock', 'SELL', None, spot, 'Scheduled year-end liquidation'); stock_qty = 0
        marks = {}
        liquidation_cost = 0.
        for name, contract in [('long', long), ('short', short)]:
            if contract is None:
                marks[name] = 0.; continue
            q = quote(day, contract['id'])
            if q is None:
                raise ValueError(f'Missing held-contract mark: {day.date()} {contract["id"]}; no forward-fill')
            marks[name] = 100 * float(q['mid']) * (1 if name == 'long' else -1)
            # Independent of execution scenario: liquidation always uses full bid/ask.
            liquidation_cost += 100 * float(q.ASK-q.BID)/2 + p.option_fee
        nav = cash+marks['long']+marks['short']+stock_qty*spot
        peak = max(peak, nav)
        ledger.append({'date': str(day.date()), 'signalDate': str(signal.date()), 'cash': round(cash, 4),
                       'longValue': round(marks['long'], 4), 'shortValue': round(marks['short'], 4),
                       'stockValue': round(stock_qty*spot, 4), 'nav': round(nav, 4),
                       'liquidationNav': round(nav-liquidation_cost-stock_qty*p.stock_half_spread, 4),
                       'drawdown': round((nav/peak-1)*100, 4), 'longContract': long['id'] if long else 'Flat',
                       'shortContract': short['id'] if short else 'Flat', 'feesToDate': round(fees, 2)})
        if reasons: decisions.append({'date': str(day.date()), 'signalDate': str(signal.date()), 'actions': reasons})
    pnl = cash-p.starting_cash
    monthly, previous = [], p.starting_cash
    for month, group in pd.DataFrame(ledger).groupby(lambda i: ledger[i]['date'][:7]):
        ending = float(group.iloc[-1]['nav'])
        monthly.append({'month': month, 'pnl': round(ending-previous, 2), 'returnPct': round((ending/previous-1)*100, 4)})
        previous = ending
    metrics = {'endingNav': round(cash, 2), 'netPnl': round(pnl, 2), 'returnPct': round(pnl/p.starting_cash*100, 4),
               'firstDebit': round(first_debit, 2), 'maxDrawdownPct': min(r['drawdown'] for r in ledger),
               'fees': round(fees, 2), 'friction': round(friction, 2), 'grossBeforeMeasuredCosts': round(pnl+fees+friction, 2),
               'oneSidedFills': one_sided_fills,
               'longLegPnl': round(flows['long'], 2), 'shortLegPnl': round(flows['short'], 2), 'stockLegPnl': round(flows['stock'], 2),
               'shortCycles': len(outcomes), 'shortWinPct': round(100*sum(o['pnl']>0 for o in outcomes)/len(outcomes), 2) if outcomes else None,
               'longRolls': long_rolls, 'fills': len(fills), 'minimumCash': min(r['cash'] for r in ledger),
               'cashReconciliationError': round(cash-p.starting_cash-sum(f['cashFlow'] for f in fills), 8),
               'navIdentityMaxError': max(abs(r['nav']-r['cash']-r['longValue']-r['shortValue']-r['stockValue']) for r in ledger),
               'skippedEntries': sum(any('Skip:' in a for a in d['actions']) for d in decisions)}
    return {'kind': kind, 'policy': asdict(p), 'metrics': metrics, 'nav': ledger, 'blotter': fills,
            'decisions': decisions, 'shortOutcomes': outcomes, 'monthly': monthly}


def export(directory, output, dsn, stock_csv=None):
    quotes, manifests = read_quotes(directory)
    stock = pd.read_csv(stock_csv) if stock_csv else read_stock(dsn)
    results = {kind: replay(stock, quotes, kind) for kind in ('pmcc', 'covered', 'long', 'stock')}
    sensitivity = []
    for delta in (.20, .25, .30):
        for fraction in (0., .5, 1.):
            run = replay(stock, quotes, p=replace(Policy(), short_delta=delta, execution_fraction=fraction))
            sensitivity.append({'shortDelta': delta, 'executionFraction': fraction, **run['metrics']})
    valid = quotes[quotes.valid]
    matched = valid[(valid['TRDPRC_1'] > 0) & valid.mid.notna()]
    # Quote/trade diagnostic uses PMCC traded contracts only, not all chains.
    traded_ids = {f['contract'] for f in results['pmcc']['blotter'] if f['leg'] != 'stock'}
    matched = matched[matched.ric.isin(traded_ids) & matched.date.between(START, END)]
    correlation = float(matched.mid.corr(matched.TRDPRC_1))
    payload = {'ticker': 'AMZN', 'start': START, 'end': END, 'startingCash': Policy().starting_cash,
               'mode': 'OBSERVED OPTION QUOTES · SIMULATED FILLS', 'sessions': len(results['pmcc']['nav']),
               'inputRows': len(stock), 'inputSha256': hashlib.sha256(stock.to_csv(index=False, float_format='%.8f').encode()).hexdigest(),
               'source': 'LSEG Workspace option bid/ask, Delta and OI; Duke WRDS / OptionMetrics stock closes',
               'results': results, 'sensitivity': sensitivity,
               'coverage': {'requestedRics': len(set(r for m in manifests for r in m['request']['rics'])),
                            'observedRics': int(valid.ric.nunique()), 'validQuoteDays': len(valid),
                            'longQuoteDays': int((valid.dte>=270).sum()), 'minDate': str(valid.date.min().date()),
                            'maxDate': str(valid.date.max().date()), 'heldMissingMarks': 0},
               'diagnostics': {'midTradePairs': len(matched), 'correlationSquared': correlation**2,
                               'meaning': 'Squared Pearson correlation, equivalent to one-regressor OLS R-squared with an intercept. Not a fill probability or evidence that mids were executable.'},
               'inputManifest': manifests, 'environment': {'python': platform.python_version(), 'pandas': pd.__version__, 'numpy': np.__version__},
               'limitations': ['EOD snapshots do not establish quote synchronization, freshness or displayed size.',
                               'Simulated next-session EOD fills, not actual executions or intraday stop fills.',
                               'No historical assignment records; early exercise is not simulated. Long call is not automatically exercised.',
                               'One stock and one fixed year; not an untouched holdout or evidence of general outperformance.',
                               'January long expiries, monthly short expiries and a bounded strike universe, not a complete option chain.']}
    output.mkdir(parents=True, exist_ok=True)
    (output/'pmcc-results.json').write_text(json.dumps(payload, separators=(',', ':'), allow_nan=False)+'\n')
    for label, rows in [('blotter', results['pmcc']['blotter']), ('ledger', results['pmcc']['nav']), ('monthly', results['pmcc']['monthly']), ('sensitivity', sensitivity)]:
        pd.DataFrame(rows).to_csv(output/f'pmcc-{label}.csv', index=False)
    shutil.copyfile(__file__, output/'pmcc_observed.py')
    shutil.copyfile(Path(__file__).with_name('README.md'), output/'README.md')
    # The project repository may be private. Share only the reproducible study
    # code and derived outputs, never database credentials or raw price chains.
    members = ['homework2/__init__.py', 'homework2/pmcc_observed.py', 'homework2/README.md',
               'homework1/__init__.py', 'homework1/option_surface_utils.py',
               'scripts/fetch_pmcc_lseg.py', 'tests/test_pmcc_observed.py', 'tests/test_pmcc_public_book.py']
    with zipfile.ZipFile(output/'pmcc-source.zip', 'w', zipfile.ZIP_DEFLATED) as archive:
        for member in members:
            path=ROOT/member
            if path.exists(): archive.write(path, member)
        for path in sorted(output.glob('pmcc-*.csv')):
            archive.write(path, 'frontend/public/pmcc/'+path.name)
        archive.write(output/'pmcc-results.json', 'frontend/public/pmcc/pmcc-results.json')
        archive.writestr('tests/__init__.py', '')
        archive.writestr('requirements.txt', f'pandas=={pd.__version__}\nnumpy=={np.__version__}\npsycopg[binary]=={psycopg.__version__}\nlseg-data==2.1.1\n')
    for kind, run in results.items(): print(kind, json.dumps(run['metrics']))
    print('COVERAGE', json.dumps(payload['coverage']))


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--data', type=Path, default=DATA)
    parser.add_argument('--output', type=Path, default=ROOT/'frontend/public/pmcc')
    parser.add_argument('--dsn', default='postgresql://localhost/duke535_research')
    parser.add_argument('--stock-csv', type=Path)
    args = parser.parse_args()
    export(args.data, args.output, args.dsn, args.stock_csv)