jonathan/Ten31-Portal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Issue 10: confirm schedule importer against real Carta XLSX

Browse Source 
- Parse Carta XLSX natively (openpyxl) instead of CSV-only
- Column map confirmed against Fund I export: Investment/Asset/
  Investment date/Shares/Cost/Value/Last Valuation date
- Handle grouped layout: company subtotal rows + indented position rows
- SAFEs: shares=0 in source → null in DB (not additive)
- Money already in dollars in XLSX cells, convert to cents directly
- Dates parsed from Excel datetime objects
- CSV fallback preserved for other formats
- Re-import guard catches both seed and non-seed existing rounds
- Tested: 14 holdings, 17 positions, total value $8,527,204.58 matches
This commit is contained in:
Johnny 5
2026-06-07 19:30:12 +00:00
parent 89eddd2ad0
commit 48fef14bc9
2 changed files with 283 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/ten31portal/routers/import_router.py
+282 -112
View File
@@ -1,4 +1,4 @@
"""CSV import endpoints for entities and schedule of investments."""
"""CSV/XLSX import endpoints for entities and schedule of investments."""
import csv
import io
@@ -6,6 +6,7 @@ import re
from datetime import date, datetime
from typing import Any
import openpyxl
from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
from sqlmodel import Session, select
@@ -21,12 +22,11 @@ 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.
# Confirmed against real Carta exports 2026-06-07.
# Entity CSV import (Issue 9) — still needs a real entity-level export to confirm.
ENTITY_COLUMN_MAP: dict[str, str] = {
# Carta export header -> model field
# UNCONFIRMED: update these after inspecting a real Carta entities export
# UNCONFIRMED: update after inspecting a real Carta entities export
"Entity Name": "name",
"Entity Type": "type",
"Vintage Year": "vintage_year",
@@ -45,14 +45,43 @@ ENTITY_TYPE_MAP: dict[str, EntityType] = {
"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",
# Schedule of Investments XLSX import (Issue 10)
# CONFIRMED against Carta export: "Low Time Preference Fund I, LLC" 2026-06-07
#
# Carta XLSX layout:
# Row 1: Entity name (e.g. "Low Time Preference Fund I, LLC")
# Row 2: Metadata line ("As of MM/DD/YYYY • Generated by ...")
# Row 3: Empty
# Row 4: Headers
# Row 5: Empty
# Rows 6+: Data (company rows alternate with position rows, separated by empty rows)
# Last data row: "Total" summary
#
# Column mapping (0-indexed from Row 4 headers):
# A (0): "Investment" — company name on company/subtotal rows
# B (1): "Asset" — security name on position rows
# C (2): "Investment date" — datetime on position rows
# D (3): "Shares" — number (0 for SAFEs/membership interests)
# E (4): "Cost" — dollar amount (float)
# F (5): "Value" — dollar amount (float)
# G (6): "Last Valuation date" — datetime on position rows
# H (7): "Gain/Loss" — derived, skip
# I (8): "Cost per share" — derived, skip
# J (9): "FMV per share" — derived, skip
# K (10): "Percent of partners' capital" — skip (phase 2)
#
# Company rows: A has name, B is empty, D/E/F have subtotals
# Position rows: A is empty, B has security name, all columns populated
# SAFEs: shares = 0, cost per share = 0, FMV per share = 0
SCHEDULE_COLUMNS = {
"investment": 0, # A — company name
"asset": 1, # B — security name
"inv_date": 2, # C — investment date
"shares": 3, # D — share count
"cost": 4, # E — cost in dollars
"value": 5, # F — value in dollars
"val_date": 6, # G — last valuation date
}
@@ -87,19 +116,12 @@ def _parse_date(raw: str) -> date | None:
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
def _dollars_to_cents(val: float | int) -> int:
"""Convert dollar amount to integer cents."""
return round(float(val) * 100)
# --- Entity import ---
# --- Entity import (CSV) ---
@router.post("/entities")
def import_entities(
@@ -116,24 +138,19 @@ def import_entities(
created = 0
updated = 0
for i, row in enumerate(reader, start=2): # row 1 is header
for i, row in enumerate(reader, start=2):
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] = []
@@ -172,7 +189,6 @@ def import_entities(
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"
@@ -219,7 +235,121 @@ def import_entities(
}
# --- Schedule of investments import ---
# --- Schedule of investments import (XLSX) ---
def _parse_schedule_xlsx(file_bytes: bytes) -> tuple[str | None, list[dict], list[dict], list[dict]]:
"""
Parse a Carta Schedule of Investments XLSX export.
Returns: (entity_name, holdings_preview, positions_preview, errors)
"""
wb = openpyxl.load_workbook(io.BytesIO(file_bytes), data_only=True)
ws = wb.active
entity_name: str | None = None
holdings_preview: list[dict] = []
positions_preview: list[dict] = []
errors: list[dict] = []
# Row 1: entity name
row1_val = ws.cell(row=1, column=1).value
if row1_val:
entity_name = str(row1_val).strip()
# Row 4: headers — validate
expected_headers = {0: "Investment", 1: "Asset", 4: "Cost", 5: "Value"}
for col_idx, expected in expected_headers.items():
actual = ws.cell(row=4, column=col_idx + 1).value
if actual and str(actual).strip() != expected:
errors.append({
"row": 4,
"errors": [f"Expected header '{expected}' in column {chr(65 + col_idx)}, got '{actual}'"],
})
# Parse data rows (starting at row 6)
current_company: str | None = None
seen_companies: set[str] = set()
for row_idx in range(6, ws.max_row + 1):
col_a = ws.cell(row=row_idx, column=1).value # Investment (company)
col_b = ws.cell(row=row_idx, column=2).value # Asset (security)
col_c = ws.cell(row=row_idx, column=3).value # Investment date
col_d = ws.cell(row=row_idx, column=4).value # Shares
col_e = ws.cell(row=row_idx, column=5).value # Cost
col_f = ws.cell(row=row_idx, column=6).value # Value
col_g = ws.cell(row=row_idx, column=7).value # Last valuation date
# Skip empty rows
if col_a is None and col_b is None:
continue
# Skip total row
if col_a and str(col_a).strip().lower() == "total":
continue
# Company subtotal row: col A has name, col B is empty
if col_a and not col_b:
company_name = str(col_a).strip()
current_company = company_name
if company_name not in seen_companies:
holdings_preview.append({
"row": row_idx,
"company_name": company_name,
})
seen_companies.add(company_name)
continue
# Position row: col B has security name
if col_b:
security_name = str(col_b).strip()
company = current_company or "(unknown)"
# Parse investment date
inv_date: date | None = None
if isinstance(col_c, datetime):
inv_date = col_c.date()
elif col_c:
inv_date = _parse_date(str(col_c))
# Parse shares (could be 0 for SAFEs)
shares: str | None = None
if col_d is not None:
shares_val = float(col_d)
if shares_val != 0:
# Preserve precision: use string repr
shares = str(col_d) if not isinstance(col_d, float) else f"{col_d:g}"
# shares stays None for zero (SAFEs)
# Parse cost (dollars)
cost_cents: int | None = None
if col_e is not None:
cost_cents = _dollars_to_cents(col_e)
# Parse value (dollars)
value_cents: int | None = None
if col_f is not None:
value_cents = _dollars_to_cents(col_f)
# Parse valuation date
val_date: date | None = None
if isinstance(col_g, datetime):
val_date = col_g.date()
elif col_g:
val_date = _parse_date(str(col_g))
positions_preview.append({
"row": row_idx,
"company_name": company,
"security_name": security_name,
"investment_date": str(inv_date) if inv_date else None,
"shares": shares,
"cost_cents": cost_cents,
"value_cents": value_cents,
"valuation_date": str(val_date) if val_date else None,
})
return entity_name, holdings_preview, positions_preview, errors
@router.post("/schedule")
def import_schedule(
@@ -234,100 +364,35 @@ def import_schedule(
if entity is None:
raise HTTPException(status_code=404, detail="Entity not found")
# Check for existing non-seed round at this quarter
# Check for any existing 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:
kind = "seed" if existing_round.is_seed else "valuation"
raise HTTPException(
status_code=409,
detail="A non-seed valuation round already exists for this entity and quarter.",
detail=f"A {kind} round already exists for this entity and quarter ({as_of}). Cannot overwrite signed history.",
)
content = file.file.read().decode("utf-8-sig")
reader = csv.DictReader(io.StringIO(content))
file_bytes = file.file.read()
filename = file.filename or ""
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,
})
# Detect format
if filename.endswith(".xlsx") or filename.endswith(".xls"):
source_entity_name, holdings_preview, positions_preview, errors = _parse_schedule_xlsx(file_bytes)
else:
# CSV fallback
source_entity_name = None
holdings_preview, positions_preview, errors = _parse_schedule_csv(file_bytes)
if not commit:
return {
"committed": False,
"source_entity_name": source_entity_name,
"holdings": holdings_preview,
"positions": positions_preview,
"errors": errors,
@@ -366,11 +431,11 @@ def import_schedule(
session.flush()
positions_created = 0
positions_updated = 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,
@@ -412,6 +477,7 @@ def import_schedule(
pos.cost_cents = pp["cost_cents"]
session.add(pos)
session.flush()
positions_updated += 1
# Attach valuation to seed round
val = Valuation(
@@ -422,16 +488,120 @@ def import_schedule(
session.add(val)
record_audit(session, user.id, "import_schedule", "entity", entity_id, {
"source_entity_name": source_entity_name,
"holdings": len(holding_map),
"positions": positions_created,
"positions_created": positions_created,
"positions_updated": positions_updated,
"seed_quarter": str(as_of),
})
session.commit()
return {
"committed": True,
"source_entity_name": source_entity_name,
"holdings_count": len(holding_map),
"positions_count": len(positions_preview),
"positions_created": positions_created,
"positions_updated": positions_updated,
"seed_round_id": seed_round.id,
"errors": errors,
}
def _parse_schedule_csv(file_bytes: bytes) -> tuple[list[dict], list[dict], list[dict]]:
"""Legacy CSV parser fallback."""
content = file_bytes.decode("utf-8-sig")
reader = csv.DictReader(io.StringIO(content))
CSV_COLUMN_MAP = {
"Company": "company_name",
"Investment": "company_name",
"Security": "security_name",
"Asset": "security_name",
"Investment Date": "investment_date",
"Investment date": "investment_date",
"Shares": "shares",
"Cost": "cost_cents",
"Value": "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:
mapped[model_field] = val.strip()
company = mapped.get("company_name", "").strip()
security = mapped.get("security_name", "").strip()
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