Ten31-Portal/backend/ten31portal/routers/import_router.py

"""XLSX/CSV import of a fund's holdings and NAV from a fund-administrator eNAV pack.

Reads the "HLD" (Holdings Report) sheet of an administrator eNAV workbook: each
security row becomes a holding + position, and its market value (book) becomes the
valuation for the quarter. Encrypted workbooks are decrypted with the open password.
A plain holdings CSV is also accepted.
"""

import csv
import io
from datetime import date, datetime
from typing import Any

import msoffcrypto
import openpyxl
from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile
from sqlmodel import Session, select, col

from ten31portal import storage
from ten31portal.audit import record_audit
from ten31portal.auth import check_administrator_scope, require_admin, require_role
from ten31portal.database import get_session
from ten31portal.models import (
    BtcPrice, Entity, EntityType, Holding, Position,
    User, UserRole, Valuation, ValuationRound, RoundStatus,
)
from ten31portal.schemas import BtcPricesImportResult, BtcPricesStatus

router = APIRouter(prefix="/api/import", tags=["import"])




def _parse_money(raw: str) -> int | None:
    """Parse dollar strings like '$3.3M', '$61.9M', '$1,234,567', '3300000' to cents."""
    if not raw or not raw.strip():
        return None
    s = raw.strip().replace(",", "").replace("$", "")
    multiplier = 1
    if s.upper().endswith("M"):
        multiplier = 1_000_000
        s = s[:-1]
    elif s.upper().endswith("K"):
        multiplier = 1_000
        s = s[:-1]
    elif s.upper().endswith("B"):
        multiplier = 1_000_000_000
        s = s[:-1]
    try:
        return round(float(s) * multiplier * 100)
    except ValueError:
        return None


def _parse_date(raw: str) -> date | None:
    """Try common date formats."""
    for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%m/%d/%y", "%Y/%m/%d"):
        try:
            return datetime.strptime(raw.strip(), fmt).date()
        except ValueError:
            continue
    return None


def _dollars_to_cents(val: float | int) -> int:
    """Convert dollar amount to integer cents."""
    return round(float(val) * 100)


# --- eNAV holdings import (XLSX) ---

def _open_workbook(file_bytes: bytes, password: str | None):
    """Load an xlsx workbook, decrypting an encrypted (password-protected) file if needed."""
    # A normal .xlsx is a zip ("PK"); an encrypted Office file is an OLE2 container.
    if file_bytes[:2] == b"PK":
        return openpyxl.load_workbook(io.BytesIO(file_bytes), data_only=True)
    try:
        office = msoffcrypto.OfficeFile(io.BytesIO(file_bytes))
        office.load_key(password=password or "VelvetSweatshop")
        out = io.BytesIO()
        office.decrypt(out)
        out.seek(0)
        return openpyxl.load_workbook(out, data_only=True)
    except Exception:
        if not password:
            raise HTTPException(
                status_code=400,
                detail="This spreadsheet is password-protected. Enter the open password and try again.",
            )
        raise HTTPException(
            status_code=400,
            detail="Could not open the spreadsheet. The password may be incorrect.",
        )


def _enav_as_of(wb) -> date | None:
    """Find the report date, e.g. a 'DECEMBER 31, 2025' / 'AS OF ...' cell near the top."""
    for sheet in (["MENU", "HLD"] if "MENU" in wb.sheetnames else wb.sheetnames):
        ws = wb[sheet]
        for row in ws.iter_rows(min_row=1, max_row=6, max_col=2, values_only=True):
            for cell in row:
                if isinstance(cell, datetime):
                    return cell.date()
                if isinstance(cell, date):
                    return cell
                if isinstance(cell, str):
                    text = cell.replace("AS OF", "").strip()
                    for fmt in ("%B %d, %Y", "%b %d, %Y", "%m/%d/%Y"):
                        try:
                            return datetime.strptime(text, fmt).date()
                        except ValueError:
                            continue
    return None


def _parse_schedule_xlsx(file_bytes: bytes, password: str | None = None) -> tuple[str | None, date | None, list[dict], list[dict], list[dict]]:
    """
    Parse the HLD (Holdings Report) sheet of a fund-administrator eNAV workbook.

    Each security row becomes a holding (issuer) + position (security), with cost basis
    and market value (book). Returns: (entity_name, as_of, holdings, positions, errors).
    """
    wb = _open_workbook(file_bytes, password)

    entity_name: str | None = None
    holdings_preview: list[dict] = []
    positions_preview: list[dict] = []
    errors: list[dict] = []

    # Fund name from the MENU cover sheet (row 2) when present.
    if "MENU" in wb.sheetnames:
        v = wb["MENU"].cell(row=2, column=1).value
        if v:
            entity_name = str(v).strip()

    as_of = _enav_as_of(wb)

    if "HLD" not in wb.sheetnames:
        errors.append({"row": 0, "errors": [
            "No 'HLD' (Holdings Report) sheet found. This does not look like an eNAV workbook."
        ]})
        return entity_name, as_of, holdings_preview, positions_preview, errors

    ws = wb["HLD"]
    rows = list(ws.iter_rows(min_row=1, max_row=ws.max_row, max_col=20, values_only=True))

    # Locate the header row (the one whose first cell is "SECURITY NAME").
    header_idx = None
    for i, row in enumerate(rows):
        if row and isinstance(row[0], str) and row[0].strip().upper() == "SECURITY NAME":
            header_idx = i
            break
    if header_idx is None:
        errors.append({"row": 0, "errors": ["Could not find the holdings header row in the HLD sheet."]})
        return entity_name, as_of, holdings_preview, positions_preview, errors

    header = [str(c).strip().upper() if isinstance(c, str) else "" for c in rows[header_idx]]

    def col(*names: str) -> int | None:
        for n in names:
            if n in header:
                return header.index(n)
        return None

    c_name = col("SECURITY NAME")
    c_qty = col("QUANTITY")
    c_cost = col("COST BASIS - BOOK", "COST BASIS - LOCAL")
    c_value = col("MARKET VALUE (BOOK)", "MARKET VALUE (LOCAL)", "MARKET VALUE - BOOK")

    if c_value is None or c_name is None:
        errors.append({"row": header_idx + 1, "errors": [
            "HLD sheet is missing a SECURITY NAME or MARKET VALUE column."
        ]})
        return entity_name, as_of, holdings_preview, positions_preview, errors

    seen_companies: set[str] = set()
    for r in range(header_idx + 1, len(rows)):
        row = rows[r]
        raw_name = row[c_name] if c_name < len(row) else None
        if raw_name is None or not str(raw_name).strip():
            continue
        name = str(raw_name).strip()
        # Stop at total / report-total summary rows.
        if name.upper().startswith("TOTAL") or name.upper().startswith("REPORT TOTAL"):
            continue

        # Group tranches of the same issuer: holding = issuer (before " - "); position = full name.
        company = name.split(" - ")[0].strip() or name
        security = name

        if company not in seen_companies:
            holdings_preview.append({"row": r + 1, "company_name": company})
            seen_companies.add(company)

        # Quantity -> shares (skip non-numeric like "N/A").
        shares: str | None = None
        if c_qty is not None and c_qty < len(row) and row[c_qty] is not None:
            qv = row[c_qty]
            if isinstance(qv, (int, float)) and qv != 0:
                shares = f"{qv:g}"

        cost_cents = None
        if c_cost is not None and c_cost < len(row) and isinstance(row[c_cost], (int, float)):
            cost_cents = _dollars_to_cents(row[c_cost])

        value_cents = None
        if isinstance(row[c_value], (int, float)):
            value_cents = _dollars_to_cents(row[c_value])

        positions_preview.append({
            "row": r + 1,
            "company_name": company,
            "security_name": security,
            "investment_date": None,  # eNAV holdings report has no acquisition date
            "shares": shares,
            "cost_cents": cost_cents,
            "value_cents": value_cents,
            "valuation_date": str(as_of) if as_of else None,
        })

    return entity_name, as_of, holdings_preview, positions_preview, errors




def reset_entity_holdings(entity_id: int, session: Session) -> dict[str, int]:
    """Delete all holdings, positions, valuations, and rounds for one entity.

    For a clean restart — e.g. after switching the source workbook (Carta → eNAV) renamed every
    position, so the old and new rows can't be matched and both get counted. Capital-account
    statements (investor data) are left untouched. The caller should re-import afterwards.
    """
    rounds = session.exec(
        select(ValuationRound).where(ValuationRound.entity_id == entity_id)
    ).all()
    round_ids = [r.id for r in rounds]
    valuations = 0
    if round_ids:
        for v in session.exec(
            select(Valuation).where(col(Valuation.round_id).in_(round_ids))
        ).all():
            session.delete(v)
            valuations += 1

    holdings = session.exec(select(Holding).where(Holding.entity_id == entity_id)).all()
    holding_ids = [h.id for h in holdings]
    positions = 0
    if holding_ids:
        for p in session.exec(
            select(Position).where(col(Position.holding_id).in_(holding_ids))
        ).all():
            session.delete(p)
            positions += 1

    for r in rounds:
        session.delete(r)
    for h in holdings:
        session.delete(h)
    session.flush()
    return {
        "rounds": len(rounds),
        "holdings": len(holdings),
        "positions": positions,
        "valuations": valuations,
    }


def dedupe_entity(entity_id: int, session: Session) -> dict[str, int]:
    """Collapse exact-duplicate holdings and positions for one entity.

    Repeated imports on older builds created a second copy of each holding/position, which
    inflated the entity's "Invested" total (the rollup sums cost across every position). This
    keeps the lowest-id copy, re-points its positions/valuations, removes duplicate valuations
    in the same round, and deletes the leftovers. Safe to run repeatedly (a no-op once clean).
    """
    removed_holdings = 0
    removed_positions = 0

    # 1) Merge holdings with the same name (case-insensitive) into the lowest-id one.
    holdings = session.exec(
        select(Holding).where(Holding.entity_id == entity_id).order_by(Holding.id)  # type: ignore[arg-type]
    ).all()
    keep_by_name: dict[str, Holding] = {}
    for h in holdings:
        key = h.company_name.strip().lower()
        keeper = keep_by_name.get(key)
        if keeper is None:
            keep_by_name[key] = h
            continue
        for p in session.exec(select(Position).where(Position.holding_id == h.id)).all():
            p.holding_id = keeper.id
            session.add(p)
        session.flush()
        session.delete(h)
        removed_holdings += 1
    session.flush()

    # 2) Within each surviving holding, merge positions with the same security name.
    for keeper_holding in keep_by_name.values():
        positions = session.exec(
            select(Position)
            .where(Position.holding_id == keeper_holding.id)
            .order_by(Position.id)  # type: ignore[arg-type]
        ).all()
        keep_by_sec: dict[str, Position] = {}
        for p in positions:
            key = p.security_name.strip().lower()
            keeper = keep_by_sec.get(key)
            if keeper is None:
                keep_by_sec[key] = p
                continue
            # Move this duplicate's valuations onto the keeper, dropping any that would collide
            # with an existing valuation for the same round (that collision IS the double-count).
            for v in session.exec(select(Valuation).where(Valuation.position_id == p.id)).all():
                clash = session.exec(
                    select(Valuation).where(
                        Valuation.round_id == v.round_id,
                        Valuation.position_id == keeper.id,
                    )
                ).first()
                if clash is not None:
                    session.delete(v)
                else:
                    v.position_id = keeper.id
                    session.add(v)
            session.flush()
            session.delete(p)
            removed_positions += 1
    session.flush()

    return {"removed_holdings": removed_holdings, "removed_positions": removed_positions}


def upsert_history_round(
    entity_id: int, as_of: date, positions_preview: list[dict], user: User, session: Session,
) -> dict[str, Any]:
    """Record a past quarter's NAV as a valuation round WITHOUT touching current holdings.

    Backfilling an old eNAV through the normal import would regress each position's cost
    basis to the old file's values and resurrect positions the fund has since exited. This
    path instead matches the file's rows by issuer + security name against the book as it
    exists today and only writes that quarter's valuations: matched rows contribute to the
    quarter's NAV, unmatched rows are counted and reported, nothing is created or modified
    outside the round. A manually signed round for the quarter is left untouched.
    """
    existing = session.exec(
        select(ValuationRound).where(
            ValuationRound.entity_id == entity_id,
            ValuationRound.quarter_end == as_of,
        )
    ).first()
    if existing is not None and not existing.is_seed:
        return {"status": "kept-signed", "matched": 0, "unmatched": 0, "nav_cents": 0}

    # Match first, so a file with no recognizable rows never leaves an empty $0 round.
    matches: list[tuple[Position, int]] = []
    unmatched = 0
    for pp in positions_preview:
        holding = session.exec(
            select(Holding).where(
                Holding.entity_id == entity_id,
                Holding.company_name == pp["company_name"],
            )
        ).first()
        pos = None
        if holding is not None:
            pos = session.exec(
                select(Position).where(
                    Position.holding_id == holding.id,
                    Position.security_name == pp["security_name"],
                )
            ).first()
        if pos is None:
            unmatched += 1
        else:
            matches.append((pos, pp["value_cents"] or 0))

    if not matches:
        return {"status": "no-match", "matched": 0, "unmatched": unmatched, "nav_cents": 0}

    round = existing
    if round is None:
        round = ValuationRound(
            entity_id=entity_id,
            quarter_end=as_of,
            status=RoundStatus.approved,
            is_seed=True,
            approved_by=user.id,
            approved_at=datetime.utcnow(),
        )
        session.add(round)
        session.flush()

    nav_cents = 0
    seen_position_ids: set[int] = set()
    for pos, value_cents in matches:
        val = session.exec(
            select(Valuation).where(
                Valuation.round_id == round.id,
                Valuation.position_id == pos.id,
            )
        ).first()
        if val is None:
            session.add(Valuation(round_id=round.id, position_id=pos.id, value_cents=value_cents))
        else:
            val.value_cents = value_cents
            session.add(val)
        nav_cents += value_cents
        seen_position_ids.add(pos.id)

    # Re-import of the same quarter: drop valuations for rows no longer in the file.
    if existing is not None:
        stale_q = select(Valuation).where(
            Valuation.round_id == round.id,
            col(Valuation.position_id).not_in(seen_position_ids),
        )
        for stale_val in session.exec(stale_q).all():
            session.delete(stale_val)

    return {
        "status": "updated" if existing is not None else "added",
        "matched": len(matches),
        "unmatched": unmatched,
        "nav_cents": nav_cents,
        "round_id": round.id,
    }


@router.post("/schedule")
def import_schedule(
    file: UploadFile = File(...),
    as_of: date | None = Query(default=None),
    entity_id: int | None = Query(default=None),
    create_entity_type: EntityType = Query(default=EntityType.fund),
    create_vintage_year: int | None = Query(default=None),
    commit: bool = Query(default=True),
    replace_existing: bool = Query(default=False),
    password: str | None = Form(default=None),
    user: User = Depends(require_admin),
    session: Session = Depends(get_session),
) -> dict[str, Any]:
    # An Administrator imports only into a fund granted to them — creating a new entity
    # (or name-matching into an arbitrary one) is internal-only.
    if user.role == UserRole.fund_administrator:
        if entity_id is None:
            raise HTTPException(
                status_code=403, detail="Choose one of your funds to import into."
            )
        check_administrator_scope(user, entity_id, session)
    try:
        file_bytes = storage.read_capped(file)
    except storage.UploadTooLarge as exc:
        raise HTTPException(status_code=413, detail=str(exc))
    filename = file.filename or ""

    # Parse the file into holdings/positions previews.
    if filename.lower().endswith((".xlsx", ".xls")):
        source_entity_name, parsed_as_of, holdings_preview, positions_preview, errors = (
            _parse_schedule_xlsx(file_bytes, password)
        )
    else:
        source_entity_name = None
        parsed_as_of = None
        holdings_preview, positions_preview, errors = _parse_schedule_csv(file_bytes)

    # As-of date: use the explicit value, else the date read from the sheet.
    as_of = as_of or parsed_as_of
    if as_of is None:
        raise HTTPException(
            status_code=400,
            detail="Could not determine the quarter-end date. Set the as-of date and try again.",
        )

    # Resolve entity
    entity: Entity | None = None
    entity_resolution: str = "existing"  # "existing" | "matched" | "will_create"

    if entity_id is not None:
        # Explicit entity_id: use it directly
        entity = session.get(Entity, entity_id)
        if entity is None:
            raise HTTPException(status_code=404, detail="Entity not found")
        entity_resolution = "existing"
    elif source_entity_name:
        # Try matching by name from row 1
        entity = session.exec(
            select(Entity).where(Entity.name == source_entity_name)
        ).first()
        if entity:
            entity_resolution = "matched"
        else:
            entity_resolution = "will_create"
    else:
        raise HTTPException(
            status_code=400,
            detail="No entity selected and no fund name found in the file. Choose the fund to import into.",
        )

    # Dry-run: report resolution without writing
    if not commit:
        resolution_info: dict[str, Any] = {"resolution": entity_resolution}
        if entity:
            resolution_info["entity_id"] = entity.id
            resolution_info["entity_name"] = entity.name
            resolution_info["entity_type"] = entity.type.value
        else:
            resolution_info["entity_name"] = source_entity_name
            resolution_info["entity_type"] = create_entity_type.value
            resolution_info["vintage_year"] = create_vintage_year
        return {
            "committed": False,
            "source_entity_name": source_entity_name,
            "entity": resolution_info,
            "holdings": holdings_preview,
            "positions": positions_preview,
            "errors": errors,
            "seed_round": {"entity_id": entity.id if entity else None, "quarter_end": str(as_of)},
        }

    # Commit path: create entity if needed
    if entity_resolution == "will_create" and source_entity_name:
        entity = Entity(
            name=source_entity_name,
            type=create_entity_type,
            vintage_year=create_vintage_year,
        )
        session.add(entity)
        session.flush()
        record_audit(session, user.id, "create", "entity", entity.id, {
            "name": source_entity_name,
            "type": create_entity_type.value,
            "source": "schedule_import",
        })

    assert entity is not None
    resolved_entity_id = entity.id

    # Replace mode: wipe the fund's existing holdings/positions/rounds first so the file becomes
    # the single source of truth. Use this after a source change (e.g. Carta → eNAV) renamed the
    # positions, leaving un-matchable rows that inflate the totals.
    if replace_existing:
        reset_entity_holdings(resolved_entity_id, session)

    # Self-heal any duplicate holdings/positions left by imports on older builds, so this
    # import's upsert lands on a single clean copy and "Invested" stops double-counting.
    dedupe_entity(resolved_entity_id, session)

    # A file OLDER than the fund's newest round is a history backfill: record the quarter's
    # NAV against today's book without touching holdings — a full import here would regress
    # cost basis to the old file and resurrect since-exited positions. (Replace mode wiped
    # the rounds above, so an explicit replace still takes the full path.)
    newest_round = session.exec(
        select(ValuationRound)
        .where(ValuationRound.entity_id == resolved_entity_id)
        .order_by(col(ValuationRound.quarter_end).desc())
    ).first()
    if newest_round is not None and newest_round.quarter_end > as_of:
        hist = upsert_history_round(resolved_entity_id, as_of, positions_preview, user, session)
        record_audit(session, user.id, "import_history", "entity", resolved_entity_id, {
            "seed_quarter": str(as_of), **{k: v for k, v in hist.items() if k != "round_id"},
        })
        session.commit()
        return {
            "committed": True,
            "history_only": True,
            **hist,
            "source_entity_name": source_entity_name,
            "entity": {
                "resolution": entity_resolution,
                "entity_id": resolved_entity_id,
                "entity_name": entity.name,
                "entity_type": entity.type.value,
            },
            "holdings_count": 0,
            "positions_count": len(positions_preview),
            "positions_created": 0,
            "positions_updated": 0,
            "seed_quarter": str(as_of),
            "errors": errors,
        }

    # An existing round at this quarter: a NAV re-import should UPDATE it in place (refresh
    # to the latest file) instead of stacking a second round and doubling the totals. Only
    # import-created seed rounds are refreshable; a manually-signed valuation round is left
    # protected.
    existing_round = session.exec(
        select(ValuationRound).where(
            ValuationRound.entity_id == resolved_entity_id,
            ValuationRound.quarter_end == as_of,
        )
    ).first()
    reused_round = False
    seed_round: ValuationRound | None = None
    if existing_round is not None:
        if not existing_round.is_seed:
            raise HTTPException(
                status_code=409,
                detail=f"A signed valuation round already exists for this entity and quarter ({as_of}). Cannot overwrite signed history.",
            )
        seed_round = existing_round
        reused_round = True

    # Commit: create holdings, positions, seed round
    holding_map: dict[str, Holding] = {}

    for hp in holdings_preview:
        name = hp["company_name"]
        existing = session.exec(
            select(Holding).where(
                Holding.entity_id == resolved_entity_id,
                Holding.company_name == name,
            )
        ).first()
        if existing:
            holding_map[name] = existing
        else:
            h = Holding(entity_id=resolved_entity_id, company_name=name)
            session.add(h)
            session.flush()
            holding_map[name] = h

    # Create the seed round, or reuse the existing one for this quarter (re-import).
    if seed_round is None:
        seed_round = ValuationRound(
            entity_id=resolved_entity_id,
            quarter_end=as_of,
            status=RoundStatus.approved,
            is_seed=True,
            approved_by=user.id,
            approved_at=datetime.utcnow(),
        )
        session.add(seed_round)
        session.flush()

    positions_created = 0
    positions_updated = 0
    seen_position_ids: set[int] = set()
    for pp in positions_preview:
        company = pp["company_name"]
        holding = holding_map.get(company)
        if holding is None:
            existing = session.exec(
                select(Holding).where(
                    Holding.entity_id == resolved_entity_id,
                    Holding.company_name == company,
                )
            ).first()
            if existing:
                holding = existing
            else:
                holding = Holding(entity_id=resolved_entity_id, company_name=company)
                session.add(holding)
                session.flush()
            holding_map[company] = holding

        # Upsert position by (holding, security_name)
        pos = session.exec(
            select(Position).where(
                Position.holding_id == holding.id,
                Position.security_name == pp["security_name"],
            )
        ).first()
        if pos is None:
            pos = Position(
                holding_id=holding.id,
                security_name=pp["security_name"],
                investment_date=_parse_date(pp["investment_date"]) if pp["investment_date"] else as_of,
                shares=pp["shares"],
                cost_cents=pp["cost_cents"] or 0,
            )
            session.add(pos)
            session.flush()
            positions_created += 1
        else:
            if pp["investment_date"]:
                pos.investment_date = _parse_date(pp["investment_date"])
            if pp["shares"]:
                pos.shares = pp["shares"]
            if pp["cost_cents"] is not None:
                pos.cost_cents = pp["cost_cents"]
            session.add(pos)
            session.flush()
            positions_updated += 1

        # Upsert this position's valuation in the round, so a re-import refreshes the value
        # in place instead of adding a second one (which would double the quarter's NAV).
        val = session.exec(
            select(Valuation).where(
                Valuation.round_id == seed_round.id,
                Valuation.position_id == pos.id,
            )
        ).first()
        if val is None:
            session.add(Valuation(
                round_id=seed_round.id,
                position_id=pos.id,
                value_cents=pp["value_cents"] or 0,
            ))
        else:
            val.value_cents = pp["value_cents"] or 0
            session.add(val)
        seen_position_ids.add(pos.id)

    # On re-import, drop valuations for holdings no longer in the file so the quarter's NAV
    # equals the new file's total (no leftovers from the prior import).
    if reused_round:
        stale_q = select(Valuation).where(Valuation.round_id == seed_round.id)
        if seen_position_ids:
            stale_q = stale_q.where(col(Valuation.position_id).not_in(seen_position_ids))
        for stale_val in session.exec(stale_q).all():
            session.delete(stale_val)

    record_audit(session, user.id, "import_schedule", "entity", resolved_entity_id, {
        "source_entity_name": source_entity_name,
        "entity_resolution": entity_resolution,
        "replaced_existing": replace_existing,
        "round_updated": reused_round,
        "holdings": len(holding_map),
        "positions_created": positions_created,
        "positions_updated": positions_updated,
        "seed_quarter": str(as_of),
    })
    session.commit()

    return {
        "committed": True,
        "source_entity_name": source_entity_name,
        "entity": {
            "resolution": entity_resolution,
            "entity_id": resolved_entity_id,
            "entity_name": entity.name,
            "entity_type": entity.type.value,
        },
        "holdings_count": len(holding_map),
        "positions_count": len(positions_preview),
        "positions_created": positions_created,
        "positions_updated": positions_updated,
        "seed_round_id": seed_round.id,
        "round_updated": reused_round,
        "replaced_existing": replace_existing,
        "errors": errors,
    }


def _parse_schedule_csv(file_bytes: bytes) -> tuple[list[dict], list[dict], list[dict]]:
    """Parse a plain holdings CSV (generic columns or an eNAV HLD export)."""
    content = file_bytes.decode("utf-8-sig")
    reader = csv.DictReader(io.StringIO(content))

    CSV_COLUMN_MAP = {
        "Company": "company_name",
        "Investment": "company_name",
        "Issuer": "company_name",
        "Security": "security_name",
        "Asset": "security_name",
        "Security Name": "security_name",
        "Investment Date": "investment_date",
        "Investment date": "investment_date",
        "Shares": "shares",
        "Quantity": "shares",
        "Cost": "cost_cents",
        "Cost Basis - Book": "cost_cents",
        "Cost Basis - Local": "cost_cents",
        "Value": "value_cents",
        "Market Value (Book)": "value_cents",
        "Market Value (Local)": "value_cents",
        "Market Value - Book": "value_cents",
    }

    holdings_preview: list[dict] = []
    positions_preview: list[dict] = []
    errors: list[dict] = []
    current_company: str | None = None

    for i, row in enumerate(reader, start=2):
        mapped: dict[str, Any] = {}
        for csv_col, model_field in CSV_COLUMN_MAP.items():
            val = row.get(csv_col)
            if val is None:
                for k, v in row.items():
                    if k.strip().lower() == csv_col.lower():
                        val = v
                        break
            if val is not None and model_field not in mapped:
                mapped[model_field] = (val or "").strip()

        company = mapped.get("company_name", "").strip()
        security = mapped.get("security_name", "").strip()
        # Skip total / summary rows.
        if security.upper().startswith("TOTAL") or company.upper().startswith("TOTAL"):
            continue
        # eNAV-style rows have only a security name; derive the issuer from its prefix.
        if security and not company:
            company = security.split(" - ")[0].strip()
            mapped["company_name"] = company

        if company and not security:
            current_company = company
            holdings_preview.append({"row": i, "company_name": company})
            continue

        if not company and current_company:
            company = current_company
        elif company:
            if company not in [h["company_name"] for h in holdings_preview]:
                holdings_preview.append({"row": i, "company_name": company})
            current_company = company

        if not security:
            continue

        row_errors: list[str] = []
        inv_date = None
        if mapped.get("investment_date"):
            inv_date = _parse_date(mapped["investment_date"])
            if inv_date is None:
                row_errors.append(f"Cannot parse date: {mapped['investment_date']}")

        shares = None
        raw_shares = mapped.get("shares", "")
        if raw_shares:
            s = raw_shares.replace(",", "")
            try:
                sv = float(s)
                if sv != 0:
                    shares = s
            except ValueError:
                row_errors.append(f"Cannot parse shares: {raw_shares}")

        cost_cents = None
        if mapped.get("cost_cents"):
            cost_cents = _parse_money(mapped["cost_cents"])
            if cost_cents is None:
                row_errors.append(f"Cannot parse cost: {mapped['cost_cents']}")

        value_cents = None
        if mapped.get("value_cents"):
            value_cents = _parse_money(mapped["value_cents"])
            if value_cents is None:
                row_errors.append(f"Cannot parse value: {mapped['value_cents']}")

        if row_errors:
            errors.append({"row": i, "errors": row_errors, "raw": dict(row)})
            continue

        positions_preview.append({
            "row": i,
            "company_name": company,
            "security_name": security,
            "investment_date": str(inv_date) if inv_date else None,
            "shares": shares,
            "cost_cents": cost_cents,
            "value_cents": value_cents,
        })

    return holdings_preview, positions_preview, errors


# --- BTC prices (bitcoin-denominated view) ---

_BTC_DATE_HEADERS = ("date", "time", "day", "snapped_at")
_BTC_PRICE_HEADERS = ("close", "price", "usd", "last", "rate")

_BTC_DATE_FORMATS = ("%Y-%m-%d", "%m/%d/%Y", "%m/%d/%y", "%Y/%m/%d", "%d-%b-%Y", "%b %d, %Y")


def _parse_btc_date(raw: str) -> date | None:
    s = raw.strip().strip('"')
    if not s:
        return None
    # Timestamps like "2026-01-01 00:00:00 UTC" — the date part is enough.
    for sep in (" ", "T"):
        if sep in s and len(s.split(sep)[0]) >= 8:
            s = s.split(sep)[0]
            break
    for fmt in _BTC_DATE_FORMATS:
        try:
            return datetime.strptime(s, fmt).date()
        except ValueError:
            continue
    return None


def _parse_btc_price(raw: str) -> int | None:
    s = raw.strip().strip('"').replace("$", "").replace(",", "").replace(" ", "")
    if not s:
        return None
    try:
        value = float(s)
    except ValueError:
        return None
    if value <= 0:
        return None
    return round(value * 100)


def _btc_status(session: Session) -> dict[str, Any]:
    rows = session.exec(select(BtcPrice).order_by(col(BtcPrice.date))).all()
    latest = rows[-1] if rows else None
    return {
        "count": len(rows),
        "first_date": rows[0].date if rows else None,
        "last_date": latest.date if latest else None,
        "latest_price_cents": latest.price_cents if latest else None,
    }


@router.get("/btc-prices")
def btc_prices_status(
    user: User = Depends(require_role(UserRole.approver, UserRole.cfo, UserRole.operations)),
    session: Session = Depends(get_session),
) -> BtcPricesStatus:
    return BtcPricesStatus(**_btc_status(session))


@router.post("/btc-prices")
def import_btc_prices(
    file: UploadFile = File(...),
    user: User = Depends(require_role(UserRole.approver, UserRole.cfo, UserRole.operations)),
    session: Session = Depends(get_session),
) -> BtcPricesImportResult:
    """Upsert BTC/USD prices from a CSV (a date column and a close/price column).

    Column headers are auto-detected; a headerless two-column file works too. Re-uploading
    overwrites prices for dates already stored, so corrections are just another upload.
    """
    try:
        file_bytes = storage.read_capped(file)
    except storage.UploadTooLarge as exc:
        raise HTTPException(status_code=413, detail=str(exc))
    try:
        text = file_bytes.decode("utf-8-sig")
    except UnicodeDecodeError:
        text = file_bytes.decode("latin-1")

    rows = [r for r in csv.reader(io.StringIO(text)) if any(c.strip() for c in r)]
    if not rows:
        raise HTTPException(status_code=400, detail="The file is empty.")

    # Column detection: match headers when present, else assume date,price.
    date_idx, price_idx, start = 0, 1, 0
    header = [c.strip().lower() for c in rows[0]]
    if not (_parse_btc_date(rows[0][0]) and len(rows[0]) > 1):
        for i, name in enumerate(header):
            if any(k in name for k in _BTC_DATE_HEADERS):
                date_idx = i
                break
        for key in _BTC_PRICE_HEADERS:  # 'close' preferred over generic 'price'
            hit = next((i for i, name in enumerate(header) if key in name and i != date_idx), None)
            if hit is not None:
                price_idx = hit
                break
        start = 1

    existing = {p.date: p for p in session.exec(select(BtcPrice)).all()}
    imported = 0
    skipped = 0
    for r in rows[start:]:
        if len(r) <= max(date_idx, price_idx):
            skipped += 1
            continue
        d = _parse_btc_date(r[date_idx])
        price = _parse_btc_price(r[price_idx])
        if d is None or price is None:
            skipped += 1
            continue
        row = existing.get(d)
        if row is None:
            row = BtcPrice(date=d, price_cents=price)
            existing[d] = row
        else:
            row.price_cents = price
        session.add(row)
        imported += 1

    if imported == 0:
        raise HTTPException(
            status_code=400,
            detail="No prices found — expected a CSV with a date column and a price column.",
        )
    record_audit(session, user.id, "import_btc_prices", "btc_prices", None,
                 detail={"imported": imported, "skipped": skipped, "filename": file.filename})
    session.commit()
    return BtcPricesImportResult(imported=imported, skipped_rows=skipped, **_btc_status(session))