438 lines
14 KiB
Python
438 lines
14 KiB
Python
"""CSV import endpoints for entities and schedule of investments."""
|
|
|
|
import csv
|
|
import io
|
|
import re
|
|
from datetime import date, datetime
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
|
|
from sqlmodel import Session, select
|
|
|
|
from ten31portal.audit import record_audit
|
|
from ten31portal.auth import require_role
|
|
from ten31portal.database import get_session
|
|
from ten31portal.models import (
|
|
Entity, EntityStatus, EntityType, Holding, Position,
|
|
User, UserRole, Valuation, ValuationRound, RoundStatus,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/import", tags=["import"])
|
|
|
|
|
|
# --- Column maps ---
|
|
# These must be confirmed against real Carta CSV exports.
|
|
# Mark as UNCONFIRMED until a sample file is validated.
|
|
|
|
ENTITY_COLUMN_MAP: dict[str, str] = {
|
|
# Carta export header -> model field
|
|
# UNCONFIRMED: update these after inspecting a real Carta entities export
|
|
"Entity Name": "name",
|
|
"Entity Type": "type",
|
|
"Vintage Year": "vintage_year",
|
|
"Fund Size": "fund_size_cents",
|
|
}
|
|
|
|
ENTITY_TYPE_MAP: dict[str, EntityType] = {
|
|
"Fund": EntityType.fund,
|
|
"fund": EntityType.fund,
|
|
"SPV": EntityType.spv,
|
|
"spv": EntityType.spv,
|
|
"GP": EntityType.gp,
|
|
"gp": EntityType.gp,
|
|
"Mgmt Co": EntityType.mgmt_co,
|
|
"mgmt_co": EntityType.mgmt_co,
|
|
"Management Company": EntityType.mgmt_co,
|
|
}
|
|
|
|
SCHEDULE_COLUMN_MAP: dict[str, str] = {
|
|
# UNCONFIRMED: update after inspecting a real Carta schedule-of-investments export
|
|
"Company": "company_name",
|
|
"Security": "security_name",
|
|
"Investment Date": "investment_date",
|
|
"Shares": "shares",
|
|
"Cost": "cost_cents",
|
|
"Value": "value_cents",
|
|
}
|
|
|
|
|
|
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 _parse_shares(raw: str) -> str | None:
|
|
"""Parse share count, return as string (Decimal stored as string)."""
|
|
if not raw or not raw.strip():
|
|
return None
|
|
s = raw.strip().replace(",", "")
|
|
try:
|
|
float(s) # Validate it's numeric
|
|
return s
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
# --- Entity import ---
|
|
|
|
@router.post("/entities")
|
|
def import_entities(
|
|
file: UploadFile = File(...),
|
|
commit: bool = Query(default=True),
|
|
user: User = Depends(require_role(UserRole.approver, UserRole.cfo)),
|
|
session: Session = Depends(get_session),
|
|
) -> dict[str, Any]:
|
|
content = file.file.read().decode("utf-8-sig")
|
|
reader = csv.DictReader(io.StringIO(content))
|
|
|
|
results: list[dict] = []
|
|
errors: list[dict] = []
|
|
created = 0
|
|
updated = 0
|
|
|
|
for i, row in enumerate(reader, start=2): # row 1 is header
|
|
mapped: dict[str, Any] = {}
|
|
unmapped_cols: list[str] = []
|
|
|
|
for csv_col, model_field in ENTITY_COLUMN_MAP.items():
|
|
val = row.get(csv_col)
|
|
if val is None:
|
|
# Try case-insensitive match
|
|
for k, v in row.items():
|
|
if k.strip().lower() == csv_col.lower():
|
|
val = v
|
|
break
|
|
if val is not None:
|
|
mapped[model_field] = val.strip()
|
|
else:
|
|
unmapped_cols.append(csv_col)
|
|
|
|
# Parse fields
|
|
parsed: dict[str, Any] = {}
|
|
row_errors: list[str] = []
|
|
|
|
name = mapped.get("name")
|
|
if not name:
|
|
row_errors.append("Missing entity name")
|
|
else:
|
|
parsed["name"] = name
|
|
|
|
type_raw = mapped.get("type", "")
|
|
entity_type = ENTITY_TYPE_MAP.get(type_raw)
|
|
if entity_type is None and type_raw:
|
|
row_errors.append(f"Unknown entity type: {type_raw}")
|
|
elif entity_type:
|
|
parsed["type"] = entity_type
|
|
|
|
vy = mapped.get("vintage_year")
|
|
if vy:
|
|
try:
|
|
parsed["vintage_year"] = int(vy)
|
|
except ValueError:
|
|
row_errors.append(f"Invalid vintage year: {vy}")
|
|
|
|
fs = mapped.get("fund_size_cents")
|
|
if fs:
|
|
cents = _parse_money(fs)
|
|
if cents is None:
|
|
row_errors.append(f"Cannot parse fund size: {fs}")
|
|
else:
|
|
parsed["fund_size_cents"] = cents
|
|
|
|
if row_errors:
|
|
errors.append({"row": i, "errors": row_errors, "raw": dict(row)})
|
|
continue
|
|
|
|
if not parsed.get("name"):
|
|
continue
|
|
|
|
# Check existing
|
|
existing = session.exec(select(Entity).where(Entity.name == parsed["name"])).first()
|
|
action = "update" if existing else "create"
|
|
|
|
results.append({
|
|
"row": i,
|
|
"action": action,
|
|
"name": parsed["name"],
|
|
"type": parsed.get("type", EntityType.fund).value if parsed.get("type") else None,
|
|
"vintage_year": parsed.get("vintage_year"),
|
|
"fund_size_cents": parsed.get("fund_size_cents"),
|
|
})
|
|
|
|
if commit:
|
|
if existing:
|
|
if "type" in parsed:
|
|
existing.type = parsed["type"]
|
|
if "vintage_year" in parsed:
|
|
existing.vintage_year = parsed["vintage_year"]
|
|
if "fund_size_cents" in parsed:
|
|
existing.fund_size_cents = parsed["fund_size_cents"]
|
|
session.add(existing)
|
|
updated += 1
|
|
else:
|
|
entity = Entity(
|
|
name=parsed["name"],
|
|
type=parsed.get("type", EntityType.fund),
|
|
vintage_year=parsed.get("vintage_year"),
|
|
fund_size_cents=parsed.get("fund_size_cents"),
|
|
)
|
|
session.add(entity)
|
|
created += 1
|
|
|
|
if commit:
|
|
record_audit(session, user.id, "import_entities", "entity", None, {
|
|
"created": created, "updated": updated, "errors": len(errors),
|
|
})
|
|
session.commit()
|
|
|
|
return {
|
|
"committed": commit,
|
|
"preview": results,
|
|
"errors": errors,
|
|
"summary": {"created": created, "updated": updated, "error_rows": len(errors)},
|
|
}
|
|
|
|
|
|
# --- Schedule of investments import ---
|
|
|
|
@router.post("/schedule")
|
|
def import_schedule(
|
|
file: UploadFile = File(...),
|
|
entity_id: int = Query(...),
|
|
as_of: date = Query(...),
|
|
commit: bool = Query(default=True),
|
|
user: User = Depends(require_role(UserRole.approver, UserRole.cfo)),
|
|
session: Session = Depends(get_session),
|
|
) -> dict[str, Any]:
|
|
entity = session.get(Entity, entity_id)
|
|
if entity is None:
|
|
raise HTTPException(status_code=404, detail="Entity not found")
|
|
|
|
# Check for existing non-seed round at this quarter
|
|
existing_round = session.exec(
|
|
select(ValuationRound).where(
|
|
ValuationRound.entity_id == entity_id,
|
|
ValuationRound.quarter_end == as_of,
|
|
ValuationRound.is_seed == False,
|
|
)
|
|
).first()
|
|
if existing_round:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="A non-seed valuation round already exists for this entity and quarter.",
|
|
)
|
|
|
|
content = file.file.read().decode("utf-8-sig")
|
|
reader = csv.DictReader(io.StringIO(content))
|
|
|
|
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 SCHEDULE_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:
|
|
mapped[model_field] = val.strip()
|
|
|
|
company = mapped.get("company_name", "").strip()
|
|
security = mapped.get("security_name", "").strip()
|
|
|
|
# Grouped layout: company row has company name but may lack security detail
|
|
if company and not security:
|
|
current_company = company
|
|
holdings_preview.append({"row": i, "company_name": company})
|
|
continue
|
|
|
|
# Position row: may inherit company from grouping
|
|
if not company and current_company:
|
|
company = current_company
|
|
elif company:
|
|
# Both company and security on same row
|
|
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 # Skip empty/subtotal rows
|
|
|
|
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 = _parse_shares(mapped.get("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,
|
|
})
|
|
|
|
if not commit:
|
|
return {
|
|
"committed": False,
|
|
"holdings": holdings_preview,
|
|
"positions": positions_preview,
|
|
"errors": errors,
|
|
"seed_round": {"entity_id": entity_id, "quarter_end": str(as_of)},
|
|
}
|
|
|
|
# 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 == entity_id,
|
|
Holding.company_name == name,
|
|
)
|
|
).first()
|
|
if existing:
|
|
holding_map[name] = existing
|
|
else:
|
|
h = Holding(entity_id=entity_id, company_name=name)
|
|
session.add(h)
|
|
session.flush()
|
|
holding_map[name] = h
|
|
|
|
# Create seed round
|
|
seed_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(seed_round)
|
|
session.flush()
|
|
|
|
positions_created = 0
|
|
for pp in positions_preview:
|
|
company = pp["company_name"]
|
|
holding = holding_map.get(company)
|
|
if holding is None:
|
|
# Create holding on the fly for positions with inline company
|
|
existing = session.exec(
|
|
select(Holding).where(
|
|
Holding.entity_id == entity_id,
|
|
Holding.company_name == company,
|
|
)
|
|
).first()
|
|
if existing:
|
|
holding = existing
|
|
else:
|
|
holding = Holding(entity_id=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()
|
|
|
|
# Attach valuation to seed round
|
|
val = Valuation(
|
|
round_id=seed_round.id,
|
|
position_id=pos.id,
|
|
value_cents=pp["value_cents"] or 0,
|
|
)
|
|
session.add(val)
|
|
|
|
record_audit(session, user.id, "import_schedule", "entity", entity_id, {
|
|
"holdings": len(holding_map),
|
|
"positions": positions_created,
|
|
"seed_quarter": str(as_of),
|
|
})
|
|
session.commit()
|
|
|
|
return {
|
|
"committed": True,
|
|
"holdings_count": len(holding_map),
|
|
"positions_count": len(positions_preview),
|
|
"seed_round_id": seed_round.id,
|
|
"errors": errors,
|
|
}
|