Next-generation infrastructure for investment management

{:ok, fund} = Ledger.open("fund-ii")

fund
|> CapitalCalls.pending()
|> Enum.each(fn call ->
  call
  |> Waterfall.allocate(fund.commitments)
  |> Enum.map(&Notice.draft(&1, call.due_date))
  |> Mailroom.send(track: true)
end)

def close_quarter(fund, period) do
  with {:ok, nav} <- Valuations.settle(fund, period),
       {:ok, accounts} <- CapitalAccounts.roll(fund, period, nav)
  ,
       statements = Enum.map(accounts, &Statement.render/1),
       :ok <- Portal.publish(statements, period: period) do
    Audit.snapshot(fund, period)
  end
end

@impl true
def handle_info(:sync_custodians, state) do
  feeds = Custodians.sync(state.fund)
  diffs = reconcile(feeds, state.fund.positions)
  if diffs != [], do: Review.queue(diffs)
  Process.send_after(self(), :sync_custodians, :timer.hours(24))
  {:noreply, state}
end

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)
  {:ok, assign(socket, investor: investor, docs: docs)}
end

def reconcile(feeds, positions) do
  by_isin = Map.new(positions, &{&1.isin, &1})

  Enum.flat_map(feeds, fn feed ->
    Enum.reject(feed.lots, &matches?(&1, by_isin[&1.isin]))
  end)
end

@waterfall Waterfall.define(preferred_return: 0.08, catch_up: 1.0
  , carry: 0.2)

{:ok, fund} = Ledger.open("fund-ii")

fund
|> CapitalCalls.pending()
|> Enum.each(fn call ->
  call
  |> Waterfall.allocate(fund.commitments)
  |> Enum.map(&Notice.draft(&1, call.due_date))
  |> Mailroom.send(track: true)
end)

def close_quarter(fund, period) do
  with {:ok, nav} <- Valuations.settle(fund, period),
       {:ok, accounts} <- CapitalAccounts.roll(fund, period, nav)
  ,
       statements = Enum.map(accounts, &Statement.render/1),
       :ok <- Portal.publish(statements, period: period) do
    Audit.snapshot(fund, period)
  end
end

@impl true
def handle_info(:sync_custodians, state) do
  feeds = Custodians.sync(state.fund)
  diffs = reconcile(feeds, state.fund.positions)
  if diffs != [], do: Review.queue(diffs)
  Process.send_after(self(), :sync_custodians, :timer.hours(24))
  {:noreply, state}
end

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)
  {:ok, assign(socket, investor: investor, docs: docs)}
end

def reconcile(feeds, positions) do
  by_isin = Map.new(positions, &{&1.isin, &1})

  Enum.flat_map(feeds, fn feed ->
    Enum.reject(feed.lots, &matches?(&1, by_isin[&1.isin]))
  end)
end

@waterfall Waterfall.define(preferred_return: 0.08, catch_up: 1.0
  , carry: 0.2)

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)
  {:ok, assign(socket, investor: investor, docs: docs)}
end

def reconcile(feeds, positions) do
  by_isin = Map.new(positions, &{&1.isin, &1})

  Enum.flat_map(feeds, fn feed ->
    Enum.reject(feed.lots, &matches?(&1, by_isin[&1.isin]))
  end)
end

@waterfall Waterfall.define(preferred_return: 0.08, catch_up: 1.0
  , carry: 0.2)

defmodule Evo.Funds.CapitalAccount do
  use Ecto.Schema
  import Ecto.Changeset

  schema "capital_accounts" do
    belongs_to :fund, Evo.Funds.Fund
    belongs_to :investor, Evo.Investors.Investor
    field :period, :date
    field :opening_balance, :decimal
    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

  def changeset(account, attrs) do
    account
    |> cast(attrs, [:period, :opening_balance, :contributions, :d
  istributions, :allocated_pnl])
    |> validate_required([:period, :opening_balance])
    |> put_closing_balance()
  end

  defp put_closing_balance(changeset) do
    opening = get_field(changeset, :opening_balance)
    pnl = get_field(changeset, :allocated_pnl) || Decimal.new(0)
    contrib = get_field(changeset, :contributions)
    dist = get_field(changeset, :distributions)

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)
  {:ok, assign(socket, investor: investor, docs: docs)}
end

def reconcile(feeds, positions) do
  by_isin = Map.new(positions, &{&1.isin, &1})

  Enum.flat_map(feeds, fn feed ->
    Enum.reject(feed.lots, &matches?(&1, by_isin[&1.isin]))
  end)
end

@waterfall Waterfall.define(preferred_return: 0.08, catch_up: 1.0
  , carry: 0.2)

defmodule Evo.Funds.CapitalAccount do
  use Ecto.Schema
  import Ecto.Changeset

  schema "capital_accounts" do
    belongs_to :fund, Evo.Funds.Fund
    belongs_to :investor, Evo.Investors.Investor
    field :period, :date
    field :opening_balance, :decimal
    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

  def changeset(account, attrs) do
    account
    |> cast(attrs, [:period, :opening_balance, :contributions, :d
  istributions, :allocated_pnl])
    |> validate_required([:period, :opening_balance])
    |> put_closing_balance()
  end

  defp put_closing_balance(changeset) do
    opening = get_field(changeset, :opening_balance)
    pnl = get_field(changeset, :allocated_pnl) || Decimal.new(0)
    contrib = get_field(changeset, :contributions)
    dist = get_field(changeset, :distributions)

    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

  def changeset(account, attrs) do
    account
    |> cast(attrs, [:period, :opening_balance, :contributions, :d
  istributions, :allocated_pnl])
    |> validate_required([:period, :opening_balance])
    |> put_closing_balance()
  end

  defp put_closing_balance(changeset) do
    opening = get_field(changeset, :opening_balance)
    pnl = get_field(changeset, :allocated_pnl) || Decimal.new(0)
    contrib = get_field(changeset, :contributions)
    dist = get_field(changeset, :distributions)

    closing =
      opening
      |> Decimal.add(contrib)
      |> Decimal.sub(dist)
      |> Decimal.add(pnl)

    put_change(changeset, :closing_balance, closing)
  end
end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"document_id" => id}}) do
    document = Documents.get!(id)

    with {:ok, pages} <- Documents.pages(document),
         {:ok, fields} <- Extraction.run(pages, schema: :custodia
  n_statement),
         {:ok, record} <- Records.upsert(document, fields) do
      Lineage.attach(record, document, fields.sources)
      Phoenix.PubSub.broadcast(Evo.PubSub, "fund:" <> document.fu
  nd_id, {:record_ready, record.id})
      :ok
    end
  end
end

    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

  def changeset(account, attrs) do
    account
    |> cast(attrs, [:period, :opening_balance, :contributions, :d
  istributions, :allocated_pnl])
    |> validate_required([:period, :opening_balance])
    |> put_closing_balance()
  end

  defp put_closing_balance(changeset) do
    opening = get_field(changeset, :opening_balance)
    pnl = get_field(changeset, :allocated_pnl) || Decimal.new(0)
    contrib = get_field(changeset, :contributions)
    dist = get_field(changeset, :distributions)

    closing =
      opening
      |> Decimal.add(contrib)
      |> Decimal.sub(dist)
      |> Decimal.add(pnl)

    put_change(changeset, :closing_balance, closing)
  end
end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"document_id" => id}}) do
    document = Documents.get!(id)

    with {:ok, pages} <- Documents.pages(document),
         {:ok, fields} <- Extraction.run(pages, schema: :custodia
  n_statement),
         {:ok, record} <- Records.upsert(document, fields) do
      Lineage.attach(record, document, fields.sources)
      Phoenix.PubSub.broadcast(Evo.PubSub, "fund:" <> document.fu
  nd_id, {:record_ready, record.id})
      :ok
    end
  end
end

end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"document_id" => id}}) do
    document = Documents.get!(id)

    with {:ok, pages} <- Documents.pages(document),
         {:ok, fields} <- Extraction.run(pages, schema: :custodia
  n_statement),
         {:ok, record} <- Records.upsert(document, fields) do
      Lineage.attach(record, document, fields.sources)
      Phoenix.PubSub.broadcast(Evo.PubSub, "fund:" <> document.fu
  nd_id, {:record_ready, record.id})
      :ok
    end
  end
end

def handle_event("approve", %{"id" => id}, socket) do
  {:ok, diff} = Review.approve(id, by: socket.assigns.current_use
  r)
  {:noreply, stream_insert(socket, :diffs, diff)}
end

def handle_event("reject", %{"id" => id, "reason" => reason}, soc
  ket) do
  {:ok, diff} = Review.reject(id, reason: reason)
  {:noreply, stream_delete(socket, :diffs, diff)}
end

defp matches?(lot, nil), do: false

defp matches?(lot, position) do
  Decimal.equal?(lot.quantity, position.quantity) and
    Decimal.equal?(lot.cost_basis, position.cost_basis) and
    lot.settlement_date == position.settlement_date
end

scope "/portal", EvoWeb do
  pipe_through [:browser, :require_investor]

  live_session :portal, on_mount: {EvoWeb.Auth, :ensure_investor}
   do
    live "/", PortalLive.Index, :index
    live "/documents", PortalLive.Documents, :index
    live "/documents/:id", PortalLive.Documents, :show

end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"document_id" => id}}) do
    document = Documents.get!(id)

    with {:ok, pages} <- Documents.pages(document),
         {:ok, fields} <- Extraction.run(pages, schema: :custodia
  n_statement),
         {:ok, record} <- Records.upsert(document, fields) do
      Lineage.attach(record, document, fields.sources)
      Phoenix.PubSub.broadcast(Evo.PubSub, "fund:" <> document.fu
  nd_id, {:record_ready, record.id})
      :ok
    end
  end
end

def handle_event("approve", %{"id" => id}, socket) do
  {:ok, diff} = Review.approve(id, by: socket.assigns.current_use
  r)
  {:noreply, stream_insert(socket, :diffs, diff)}
end

def handle_event("reject", %{"id" => id, "reason" => reason}, soc
  ket) do
  {:ok, diff} = Review.reject(id, reason: reason)
  {:noreply, stream_delete(socket, :diffs, diff)}
end

defp matches?(lot, nil), do: false

defp matches?(lot, position) do
  Decimal.equal?(lot.quantity, position.quantity) and
    Decimal.equal?(lot.cost_basis, position.cost_basis) and
    lot.settlement_date == position.settlement_date
end

scope "/portal", EvoWeb do
  pipe_through [:browser, :require_investor]

  live_session :portal, on_mount: {EvoWeb.Auth, :ensure_investor}
   do
    live "/", PortalLive.Index, :index
    live "/documents", PortalLive.Documents, :index
    live "/documents/:id", PortalLive.Documents, :show

  {:ok, diff} = Review.reject(id, reason: reason)
  {:noreply, stream_delete(socket, :diffs, diff)}
end

defp matches?(lot, nil), do: false

defp matches?(lot, position) do
  Decimal.equal?(lot.quantity, position.quantity) and
    Decimal.equal?(lot.cost_basis, position.cost_basis) and
    lot.settlement_date == position.settlement_date
end

scope "/portal", EvoWeb do
  pipe_through [:browser, :require_investor]

  live_session :portal, on_mount: {EvoWeb.Auth, :ensure_investor}
   do
    live "/", PortalLive.Index, :index
    live "/documents", PortalLive.Documents, :index
    live "/documents/:id", PortalLive.Documents, :show
    live "/capital-account", PortalLive.CapitalAccount, :index
  end
end

def allocate(%Call{amount: amount} = call, commitments) do
  total = commitments |> Enum.map(& &1.amount) |> Enum.reduce(Dec
  imal.new(0), &Decimal.add/2)

  Enum.map(commitments, fn commitment ->
    share = Decimal.div(commitment.amount, total)

    %Allocation{
      investor_id: commitment.investor_id,
      call_id: call.id,
      amount: Decimal.mult(amount, share) |> Decimal.round(2),
      due_date: call.due_date
    }
  end)
end

test "closing the quarter rolls every capital account" do
  fund = insert(:fund, investors: 3)
  period = ~D[2026-06-30]

  assert {:ok, snapshot} = Funds.close_quarter(fund, period)
  assert length(snapshot.accounts) == 3
  assert Enum.all?(snapshot.accounts, &Decimal.positive?(&1.closi
  ng_balance))
end

  {:ok, diff} = Review.reject(id, reason: reason)
  {:noreply, stream_delete(socket, :diffs, diff)}
end

defp matches?(lot, nil), do: false

defp matches?(lot, position) do
  Decimal.equal?(lot.quantity, position.quantity) and
    Decimal.equal?(lot.cost_basis, position.cost_basis) and
    lot.settlement_date == position.settlement_date
end

scope "/portal", EvoWeb do
  pipe_through [:browser, :require_investor]

  live_session :portal, on_mount: {EvoWeb.Auth, :ensure_investor}
   do
    live "/", PortalLive.Index, :index
    live "/documents", PortalLive.Documents, :index
    live "/documents/:id", PortalLive.Documents, :show
    live "/capital-account", PortalLive.CapitalAccount, :index
  end
end

def allocate(%Call{amount: amount} = call, commitments) do
  total = commitments |> Enum.map(& &1.amount) |> Enum.reduce(Dec
  imal.new(0), &Decimal.add/2)

  Enum.map(commitments, fn commitment ->
    share = Decimal.div(commitment.amount, total)

    %Allocation{
      investor_id: commitment.investor_id,
      call_id: call.id,
      amount: Decimal.mult(amount, share) |> Decimal.round(2),
      due_date: call.due_date
    }
  end)
end

test "closing the quarter rolls every capital account" do
  fund = insert(:fund, investors: 3)
  period = ~D[2026-06-30]

  assert {:ok, snapshot} = Funds.close_quarter(fund, period)
  assert length(snapshot.accounts) == 3
  assert Enum.all?(snapshot.accounts, &Decimal.positive?(&1.closi
  ng_balance))
end

    share = Decimal.div(commitment.amount, total)

    %Allocation{
      investor_id: commitment.investor_id,
      call_id: call.id,
      amount: Decimal.mult(amount, share) |> Decimal.round(2),
      due_date: call.due_date
    }
  end)
end

test "closing the quarter rolls every capital account" do
  fund = insert(:fund, investors: 3)
  period = ~D[2026-06-30]

  assert {:ok, snapshot} = Funds.close_quarter(fund, period)
  assert length(snapshot.accounts) == 3
  assert Enum.all?(snapshot.accounts, &Decimal.positive?(&1.closi
  ng_balance))
end

defmodule Evo.Repo.Migrations.AddLineageToRecords do
  use Ecto.Migration

  def change do
    alter table(:records) do
      add :source_document_id, references(:documents, on_delete: 
  :nothing)
      add :source_page, :integer
      add :source_bbox, {:array, :float}
    end

    create index(:records, [:source_document_id])
  end
end

{:ok, fund} = Ledger.open("fund-ii")

fund
|> CapitalCalls.pending()
|> Enum.each(fn call ->
  call
  |> Waterfall.allocate(fund.commitments)
  |> Enum.map(&Notice.draft(&1, call.due_date))
  |> Mailroom.send(track: true)
end)

def close_quarter(fund, period) do
  with {:ok, nav} <- Valuations.settle(fund, period),

    share = Decimal.div(commitment.amount, total)

    %Allocation{
      investor_id: commitment.investor_id,
      call_id: call.id,
      amount: Decimal.mult(amount, share) |> Decimal.round(2),
      due_date: call.due_date
    }
  end)
end

test "closing the quarter rolls every capital account" do
  fund = insert(:fund, investors: 3)
  period = ~D[2026-06-30]

  assert {:ok, snapshot} = Funds.close_quarter(fund, period)
  assert length(snapshot.accounts) == 3
  assert Enum.all?(snapshot.accounts, &Decimal.positive?(&1.closi
  ng_balance))
end

defmodule Evo.Repo.Migrations.AddLineageToRecords do
  use Ecto.Migration

  def change do
    alter table(:records) do
      add :source_document_id, references(:documents, on_delete: 
  :nothing)
      add :source_page, :integer
      add :source_bbox, {:array, :float}
    end

    create index(:records, [:source_document_id])
  end
end

{:ok, fund} = Ledger.open("fund-ii")

fund
|> CapitalCalls.pending()
|> Enum.each(fn call ->
  call
  |> Waterfall.allocate(fund.commitments)
  |> Enum.map(&Notice.draft(&1, call.due_date))
  |> Mailroom.send(track: true)
end)

def close_quarter(fund, period) do
  with {:ok, nav} <- Valuations.settle(fund, period),

      add :source_bbox, {:array, :float}
    end

    create index(:records, [:source_document_id])
  end
end

{:ok, fund} = Ledger.open("fund-ii")

fund
|> CapitalCalls.pending()
|> Enum.each(fn call ->
  call
  |> Waterfall.allocate(fund.commitments)
  |> Enum.map(&Notice.draft(&1, call.due_date))
  |> Mailroom.send(track: true)
end)

def close_quarter(fund, period) do
  with {:ok, nav} <- Valuations.settle(fund, period),
       {:ok, accounts} <- CapitalAccounts.roll(fund, period, nav)
  ,
       statements = Enum.map(accounts, &Statement.render/1),
       :ok <- Portal.publish(statements, period: period) do
    Audit.snapshot(fund, period)
  end
end

@impl true
def handle_info(:sync_custodians, state) do
  feeds = Custodians.sync(state.fund)
  diffs = reconcile(feeds, state.fund.positions)
  if diffs != [], do: Review.queue(diffs)
  Process.send_after(self(), :sync_custodians, :timer.hours(24))
  {:noreply, state}
end

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)
  {:ok, assign(socket, investor: investor, docs: docs)}
end

def reconcile(feeds, positions) do
  by_isin = Map.new(positions, &{&1.isin, &1})

  Enum.flat_map(feeds, fn feed ->

      add :source_bbox, {:array, :float}
    end

    create index(:records, [:source_document_id])
  end
end

{:ok, fund} = Ledger.open("fund-ii")

fund
|> CapitalCalls.pending()
|> Enum.each(fn call ->
  call
  |> Waterfall.allocate(fund.commitments)
  |> Enum.map(&Notice.draft(&1, call.due_date))
  |> Mailroom.send(track: true)
end)

def close_quarter(fund, period) do
  with {:ok, nav} <- Valuations.settle(fund, period),
       {:ok, accounts} <- CapitalAccounts.roll(fund, period, nav)
  ,
       statements = Enum.map(accounts, &Statement.render/1),
       :ok <- Portal.publish(statements, period: period) do
    Audit.snapshot(fund, period)
  end
end

@impl true
def handle_info(:sync_custodians, state) do
  feeds = Custodians.sync(state.fund)
  diffs = reconcile(feeds, state.fund.positions)
  if diffs != [], do: Review.queue(diffs)
  Process.send_after(self(), :sync_custodians, :timer.hours(24))
  {:noreply, state}
end

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)
  {:ok, assign(socket, investor: investor, docs: docs)}
end

def reconcile(feeds, positions) do
  by_isin = Map.new(positions, &{&1.isin, &1})

  Enum.flat_map(feeds, fn feed ->

def handle_info(:sync_custodians, state) do
  feeds = Custodians.sync(state.fund)
  diffs = reconcile(feeds, state.fund.positions)
  if diffs != [], do: Review.queue(diffs)
  Process.send_after(self(), :sync_custodians, :timer.hours(24))
  {:noreply, state}
end

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)
  {:ok, assign(socket, investor: investor, docs: docs)}
end

def reconcile(feeds, positions) do
  by_isin = Map.new(positions, &{&1.isin, &1})

  Enum.flat_map(feeds, fn feed ->
    Enum.reject(feed.lots, &matches?(&1, by_isin[&1.isin]))
  end)
end

@waterfall Waterfall.define(preferred_return: 0.08, catch_up: 1.0
  , carry: 0.2)

defmodule Evo.Funds.CapitalAccount do
  use Ecto.Schema
  import Ecto.Changeset

  schema "capital_accounts" do
    belongs_to :fund, Evo.Funds.Fund
    belongs_to :investor, Evo.Investors.Investor
    field :period, :date
    field :opening_balance, :decimal
    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

  def changeset(account, attrs) do
    account
    |> cast(attrs, [:period, :opening_balance, :contributions, :d
  istributions, :allocated_pnl])
    |> validate_required([:period, :opening_balance])
    |> put_closing_balance()

def handle_info(:sync_custodians, state) do
  feeds = Custodians.sync(state.fund)
  diffs = reconcile(feeds, state.fund.positions)
  if diffs != [], do: Review.queue(diffs)
  Process.send_after(self(), :sync_custodians, :timer.hours(24))
  {:noreply, state}
end

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)
  {:ok, assign(socket, investor: investor, docs: docs)}
end

def reconcile(feeds, positions) do
  by_isin = Map.new(positions, &{&1.isin, &1})

  Enum.flat_map(feeds, fn feed ->
    Enum.reject(feed.lots, &matches?(&1, by_isin[&1.isin]))
  end)
end

@waterfall Waterfall.define(preferred_return: 0.08, catch_up: 1.0
  , carry: 0.2)

defmodule Evo.Funds.CapitalAccount do
  use Ecto.Schema
  import Ecto.Changeset

  schema "capital_accounts" do
    belongs_to :fund, Evo.Funds.Fund
    belongs_to :investor, Evo.Investors.Investor
    field :period, :date
    field :opening_balance, :decimal
    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

  def changeset(account, attrs) do
    account
    |> cast(attrs, [:period, :opening_balance, :contributions, :d
  istributions, :allocated_pnl])
    |> validate_required([:period, :opening_balance])
    |> put_closing_balance()

  import Ecto.Changeset

  schema "capital_accounts" do
    belongs_to :fund, Evo.Funds.Fund
    belongs_to :investor, Evo.Investors.Investor
    field :period, :date
    field :opening_balance, :decimal
    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

  def changeset(account, attrs) do
    account
    |> cast(attrs, [:period, :opening_balance, :contributions, :d
  istributions, :allocated_pnl])
    |> validate_required([:period, :opening_balance])
    |> put_closing_balance()
  end

  defp put_closing_balance(changeset) do
    opening = get_field(changeset, :opening_balance)
    pnl = get_field(changeset, :allocated_pnl) || Decimal.new(0)
    contrib = get_field(changeset, :contributions)
    dist = get_field(changeset, :distributions)

    closing =
      opening
      |> Decimal.add(contrib)
      |> Decimal.sub(dist)
      |> Decimal.add(pnl)

    put_change(changeset, :closing_balance, closing)
  end
end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"document_id" => id}}) do
    document = Documents.get!(id)

    with {:ok, pages} <- Documents.pages(document),
         {:ok, fields} <- Extraction.run(pages, schema: :custodia
  n_statement),
         {:ok, record} <- Records.upsert(document, fields) do

  import Ecto.Changeset

  schema "capital_accounts" do
    belongs_to :fund, Evo.Funds.Fund
    belongs_to :investor, Evo.Investors.Investor
    field :period, :date
    field :opening_balance, :decimal
    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

  def changeset(account, attrs) do
    account
    |> cast(attrs, [:period, :opening_balance, :contributions, :d
  istributions, :allocated_pnl])
    |> validate_required([:period, :opening_balance])
    |> put_closing_balance()
  end

  defp put_closing_balance(changeset) do
    opening = get_field(changeset, :opening_balance)
    pnl = get_field(changeset, :allocated_pnl) || Decimal.new(0)
    contrib = get_field(changeset, :contributions)
    dist = get_field(changeset, :distributions)

    closing =
      opening
      |> Decimal.add(contrib)
      |> Decimal.sub(dist)
      |> Decimal.add(pnl)

    put_change(changeset, :closing_balance, closing)
  end
end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"document_id" => id}}) do
    document = Documents.get!(id)

    with {:ok, pages} <- Documents.pages(document),
         {:ok, fields} <- Extraction.run(pages, schema: :custodia
  n_statement),
         {:ok, record} <- Records.upsert(document, fields) do

      opening
      |> Decimal.add(contrib)
      |> Decimal.sub(dist)
      |> Decimal.add(pnl)

    put_change(changeset, :closing_balance, closing)
  end
end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"document_id" => id}}) do
    document = Documents.get!(id)

    with {:ok, pages} <- Documents.pages(document),
         {:ok, fields} <- Extraction.run(pages, schema: :custodia
  n_statement),
         {:ok, record} <- Records.upsert(document, fields) do
      Lineage.attach(record, document, fields.sources)
      Phoenix.PubSub.broadcast(Evo.PubSub, "fund:" <> document.fu
  nd_id, {:record_ready, record.id})
      :ok
    end
  end
end

def handle_event("approve", %{"id" => id}, socket) do
  {:ok, diff} = Review.approve(id, by: socket.assigns.current_use
  r)
  {:noreply, stream_insert(socket, :diffs, diff)}
end

def handle_event("reject", %{"id" => id, "reason" => reason}, soc
  ket) do
  {:ok, diff} = Review.reject(id, reason: reason)
  {:noreply, stream_delete(socket, :diffs, diff)}
end

defp matches?(lot, nil), do: false

defp matches?(lot, position) do
  Decimal.equal?(lot.quantity, position.quantity) and
    Decimal.equal?(lot.cost_basis, position.cost_basis) and
    lot.settlement_date == position.settlement_date
end

scope "/portal", EvoWeb do

      opening
      |> Decimal.add(contrib)
      |> Decimal.sub(dist)
      |> Decimal.add(pnl)

    put_change(changeset, :closing_balance, closing)
  end
end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"document_id" => id}}) do
    document = Documents.get!(id)

    with {:ok, pages} <- Documents.pages(document),
         {:ok, fields} <- Extraction.run(pages, schema: :custodia
  n_statement),
         {:ok, record} <- Records.upsert(document, fields) do
      Lineage.attach(record, document, fields.sources)
      Phoenix.PubSub.broadcast(Evo.PubSub, "fund:" <> document.fu
  nd_id, {:record_ready, record.id})
      :ok
    end
  end
end

def handle_event("approve", %{"id" => id}, socket) do
  {:ok, diff} = Review.approve(id, by: socket.assigns.current_use
  r)
  {:noreply, stream_insert(socket, :diffs, diff)}
end

def handle_event("reject", %{"id" => id, "reason" => reason}, soc
  ket) do
  {:ok, diff} = Review.reject(id, reason: reason)
  {:noreply, stream_delete(socket, :diffs, diff)}
end

defp matches?(lot, nil), do: false

defp matches?(lot, position) do
  Decimal.equal?(lot.quantity, position.quantity) and
    Decimal.equal?(lot.cost_basis, position.cost_basis) and
    lot.settlement_date == position.settlement_date
end

scope "/portal", EvoWeb do

  {:ok, diff} = Review.approve(id, by: socket.assigns.current_use
  r)
  {:noreply, stream_insert(socket, :diffs, diff)}
end

def handle_event("reject", %{"id" => id, "reason" => reason}, soc
  ket) do
  {:ok, diff} = Review.reject(id, reason: reason)
  {:noreply, stream_delete(socket, :diffs, diff)}
end

defp matches?(lot, nil), do: false

defp matches?(lot, position) do
  Decimal.equal?(lot.quantity, position.quantity) and
    Decimal.equal?(lot.cost_basis, position.cost_basis) and
    lot.settlement_date == position.settlement_date
end

scope "/portal", EvoWeb do
  pipe_through [:browser, :require_investor]

  live_session :portal, on_mount: {EvoWeb.Auth, :ensure_investor}
   do
    live "/", PortalLive.Index, :index
    live "/documents", PortalLive.Documents, :index
    live "/documents/:id", PortalLive.Documents, :show
    live "/capital-account", PortalLive.CapitalAccount, :index
  end
end

def allocate(%Call{amount: amount} = call, commitments) do
  total = commitments |> Enum.map(& &1.amount) |> Enum.reduce(Dec
  imal.new(0), &Decimal.add/2)

  Enum.map(commitments, fn commitment ->
    share = Decimal.div(commitment.amount, total)

    %Allocation{
      investor_id: commitment.investor_id,
      call_id: call.id,
      amount: Decimal.mult(amount, share) |> Decimal.round(2),
      due_date: call.due_date
    }
  end)
end

test "closing the quarter rolls every capital account" do
  fund = insert(:fund, investors: 3)

  {:ok, diff} = Review.approve(id, by: socket.assigns.current_use
  r)
  {:noreply, stream_insert(socket, :diffs, diff)}
end

def handle_event("reject", %{"id" => id, "reason" => reason}, soc
  ket) do
  {:ok, diff} = Review.reject(id, reason: reason)
  {:noreply, stream_delete(socket, :diffs, diff)}
end

defp matches?(lot, nil), do: false

defp matches?(lot, position) do
  Decimal.equal?(lot.quantity, position.quantity) and
    Decimal.equal?(lot.cost_basis, position.cost_basis) and
    lot.settlement_date == position.settlement_date
end

scope "/portal", EvoWeb do
  pipe_through [:browser, :require_investor]

  live_session :portal, on_mount: {EvoWeb.Auth, :ensure_investor}
   do
    live "/", PortalLive.Index, :index
    live "/documents", PortalLive.Documents, :index
    live "/documents/:id", PortalLive.Documents, :show
    live "/capital-account", PortalLive.CapitalAccount, :index
  end
end

def allocate(%Call{amount: amount} = call, commitments) do
  total = commitments |> Enum.map(& &1.amount) |> Enum.reduce(Dec
  imal.new(0), &Decimal.add/2)

  Enum.map(commitments, fn commitment ->
    share = Decimal.div(commitment.amount, total)

    %Allocation{
      investor_id: commitment.investor_id,
      call_id: call.id,
      amount: Decimal.mult(amount, share) |> Decimal.round(2),
      due_date: call.due_date
    }
  end)
end

test "closing the quarter rolls every capital account" do
  fund = insert(:fund, investors: 3)

end

def allocate(%Call{amount: amount} = call, commitments) do
  total = commitments |> Enum.map(& &1.amount) |> Enum.reduce(Dec
  imal.new(0), &Decimal.add/2)

  Enum.map(commitments, fn commitment ->
    share = Decimal.div(commitment.amount, total)

    %Allocation{
      investor_id: commitment.investor_id,
      call_id: call.id,
      amount: Decimal.mult(amount, share) |> Decimal.round(2),
      due_date: call.due_date
    }
  end)
end

test "closing the quarter rolls every capital account" do
  fund = insert(:fund, investors: 3)
  period = ~D[2026-06-30]

  assert {:ok, snapshot} = Funds.close_quarter(fund, period)
  assert length(snapshot.accounts) == 3
  assert Enum.all?(snapshot.accounts, &Decimal.positive?(&1.closi
  ng_balance))
end

defmodule Evo.Repo.Migrations.AddLineageToRecords do
  use Ecto.Migration

  def change do
    alter table(:records) do
      add :source_document_id, references(:documents, on_delete: 
  :nothing)
      add :source_page, :integer
      add :source_bbox, {:array, :float}
    end

    create index(:records, [:source_document_id])
  end
end

{:ok, fund} = Ledger.open("fund-ii")

fund
|> CapitalCalls.pending()
|> Enum.each(fn call ->
  call

end

def allocate(%Call{amount: amount} = call, commitments) do
  total = commitments |> Enum.map(& &1.amount) |> Enum.reduce(Dec
  imal.new(0), &Decimal.add/2)

  Enum.map(commitments, fn commitment ->
    share = Decimal.div(commitment.amount, total)

    %Allocation{
      investor_id: commitment.investor_id,
      call_id: call.id,
      amount: Decimal.mult(amount, share) |> Decimal.round(2),
      due_date: call.due_date
    }
  end)
end

test "closing the quarter rolls every capital account" do
  fund = insert(:fund, investors: 3)
  period = ~D[2026-06-30]

  assert {:ok, snapshot} = Funds.close_quarter(fund, period)
  assert length(snapshot.accounts) == 3
  assert Enum.all?(snapshot.accounts, &Decimal.positive?(&1.closi
  ng_balance))
end

defmodule Evo.Repo.Migrations.AddLineageToRecords do
  use Ecto.Migration

  def change do
    alter table(:records) do
      add :source_document_id, references(:documents, on_delete: 
  :nothing)
      add :source_page, :integer
      add :source_bbox, {:array, :float}
    end

    create index(:records, [:source_document_id])
  end
end

{:ok, fund} = Ledger.open("fund-ii")

fund
|> CapitalCalls.pending()
|> Enum.each(fn call ->
  call

  use Ecto.Migration

  def change do
    alter table(:records) do
      add :source_document_id, references(:documents, on_delete: 
  :nothing)
      add :source_page, :integer
      add :source_bbox, {:array, :float}
    end

    create index(:records, [:source_document_id])
  end
end

{:ok, fund} = Ledger.open("fund-ii")

fund
|> CapitalCalls.pending()
|> Enum.each(fn call ->
  call
  |> Waterfall.allocate(fund.commitments)
  |> Enum.map(&Notice.draft(&1, call.due_date))
  |> Mailroom.send(track: true)
end)

def close_quarter(fund, period) do
  with {:ok, nav} <- Valuations.settle(fund, period),
       {:ok, accounts} <- CapitalAccounts.roll(fund, period, nav)
  ,
       statements = Enum.map(accounts, &Statement.render/1),
       :ok <- Portal.publish(statements, period: period) do
    Audit.snapshot(fund, period)
  end
end

@impl true
def handle_info(:sync_custodians, state) do
  feeds = Custodians.sync(state.fund)
  diffs = reconcile(feeds, state.fund.positions)
  if diffs != [], do: Review.queue(diffs)
  Process.send_after(self(), :sync_custodians, :timer.hours(24))
  {:noreply, state}
end

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)

  use Ecto.Migration

  def change do
    alter table(:records) do
      add :source_document_id, references(:documents, on_delete: 
  :nothing)
      add :source_page, :integer
      add :source_bbox, {:array, :float}
    end

    create index(:records, [:source_document_id])
  end
end

{:ok, fund} = Ledger.open("fund-ii")

fund
|> CapitalCalls.pending()
|> Enum.each(fn call ->
  call
  |> Waterfall.allocate(fund.commitments)
  |> Enum.map(&Notice.draft(&1, call.due_date))
  |> Mailroom.send(track: true)
end)

def close_quarter(fund, period) do
  with {:ok, nav} <- Valuations.settle(fund, period),
       {:ok, accounts} <- CapitalAccounts.roll(fund, period, nav)
  ,
       statements = Enum.map(accounts, &Statement.render/1),
       :ok <- Portal.publish(statements, period: period) do
    Audit.snapshot(fund, period)
  end
end

@impl true
def handle_info(:sync_custodians, state) do
  feeds = Custodians.sync(state.fund)
  diffs = reconcile(feeds, state.fund.positions)
  if diffs != [], do: Review.queue(diffs)
  Process.send_after(self(), :sync_custodians, :timer.hours(24))
  {:noreply, state}
end

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)

       statements = Enum.map(accounts, &Statement.render/1),
       :ok <- Portal.publish(statements, period: period) do
    Audit.snapshot(fund, period)
  end
end

@impl true
def handle_info(:sync_custodians, state) do
  feeds = Custodians.sync(state.fund)
  diffs = reconcile(feeds, state.fund.positions)
  if diffs != [], do: Review.queue(diffs)
  Process.send_after(self(), :sync_custodians, :timer.hours(24))
  {:noreply, state}
end

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)
  {:ok, assign(socket, investor: investor, docs: docs)}
end

def reconcile(feeds, positions) do
  by_isin = Map.new(positions, &{&1.isin, &1})

  Enum.flat_map(feeds, fn feed ->
    Enum.reject(feed.lots, &matches?(&1, by_isin[&1.isin]))
  end)
end

@waterfall Waterfall.define(preferred_return: 0.08, catch_up: 1.0
  , carry: 0.2)

defmodule Evo.Funds.CapitalAccount do
  use Ecto.Schema
  import Ecto.Changeset

  schema "capital_accounts" do
    belongs_to :fund, Evo.Funds.Fund
    belongs_to :investor, Evo.Investors.Investor
    field :period, :date
    field :opening_balance, :decimal
    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

       statements = Enum.map(accounts, &Statement.render/1),
       :ok <- Portal.publish(statements, period: period) do
    Audit.snapshot(fund, period)
  end
end

@impl true
def handle_info(:sync_custodians, state) do
  feeds = Custodians.sync(state.fund)
  diffs = reconcile(feeds, state.fund.positions)
  if diffs != [], do: Review.queue(diffs)
  Process.send_after(self(), :sync_custodians, :timer.hours(24))
  {:noreply, state}
end

@impl true
def mount(%{"email" => email}, _session, socket) do
  investor = CRM.get_by!(email: email)
  docs = Dataroom.list_for(investor, since: investor.last_login_a
  t)
  {:ok, assign(socket, investor: investor, docs: docs)}
end

def reconcile(feeds, positions) do
  by_isin = Map.new(positions, &{&1.isin, &1})

  Enum.flat_map(feeds, fn feed ->
    Enum.reject(feed.lots, &matches?(&1, by_isin[&1.isin]))
  end)
end

@waterfall Waterfall.define(preferred_return: 0.08, catch_up: 1.0
  , carry: 0.2)

defmodule Evo.Funds.CapitalAccount do
  use Ecto.Schema
  import Ecto.Changeset

  schema "capital_accounts" do
    belongs_to :fund, Evo.Funds.Fund
    belongs_to :investor, Evo.Investors.Investor
    field :period, :date
    field :opening_balance, :decimal
    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

end

@waterfall Waterfall.define(preferred_return: 0.08, catch_up: 1.0
  , carry: 0.2)

defmodule Evo.Funds.CapitalAccount do
  use Ecto.Schema
  import Ecto.Changeset

  schema "capital_accounts" do
    belongs_to :fund, Evo.Funds.Fund
    belongs_to :investor, Evo.Investors.Investor
    field :period, :date
    field :opening_balance, :decimal
    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

  def changeset(account, attrs) do
    account
    |> cast(attrs, [:period, :opening_balance, :contributions, :d
  istributions, :allocated_pnl])
    |> validate_required([:period, :opening_balance])
    |> put_closing_balance()
  end

  defp put_closing_balance(changeset) do
    opening = get_field(changeset, :opening_balance)
    pnl = get_field(changeset, :allocated_pnl) || Decimal.new(0)
    contrib = get_field(changeset, :contributions)
    dist = get_field(changeset, :distributions)

    closing =
      opening
      |> Decimal.add(contrib)
      |> Decimal.sub(dist)
      |> Decimal.add(pnl)

    put_change(changeset, :closing_balance, closing)
  end
end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker

end

@waterfall Waterfall.define(preferred_return: 0.08, catch_up: 1.0
  , carry: 0.2)

defmodule Evo.Funds.CapitalAccount do
  use Ecto.Schema
  import Ecto.Changeset

  schema "capital_accounts" do
    belongs_to :fund, Evo.Funds.Fund
    belongs_to :investor, Evo.Investors.Investor
    field :period, :date
    field :opening_balance, :decimal
    field :contributions, :decimal, default: Decimal.new(0)
    field :distributions, :decimal, default: Decimal.new(0)
    field :allocated_pnl, :decimal
    field :closing_balance, :decimal
    timestamps()
  end

  def changeset(account, attrs) do
    account
    |> cast(attrs, [:period, :opening_balance, :contributions, :d
  istributions, :allocated_pnl])
    |> validate_required([:period, :opening_balance])
    |> put_closing_balance()
  end

  defp put_closing_balance(changeset) do
    opening = get_field(changeset, :opening_balance)
    pnl = get_field(changeset, :allocated_pnl) || Decimal.new(0)
    contrib = get_field(changeset, :contributions)
    dist = get_field(changeset, :distributions)

    closing =
      opening
      |> Decimal.add(contrib)
      |> Decimal.sub(dist)
      |> Decimal.add(pnl)

    put_change(changeset, :closing_balance, closing)
  end
end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker

  defp put_closing_balance(changeset) do
    opening = get_field(changeset, :opening_balance)
    pnl = get_field(changeset, :allocated_pnl) || Decimal.new(0)
    contrib = get_field(changeset, :contributions)
    dist = get_field(changeset, :distributions)

    closing =
      opening
      |> Decimal.add(contrib)
      |> Decimal.sub(dist)
      |> Decimal.add(pnl)

    put_change(changeset, :closing_balance, closing)
  end
end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"document_id" => id}}) do
    document = Documents.get!(id)

    with {:ok, pages} <- Documents.pages(document),
         {:ok, fields} <- Extraction.run(pages, schema: :custodia
  n_statement),
         {:ok, record} <- Records.upsert(document, fields) do
      Lineage.attach(record, document, fields.sources)
      Phoenix.PubSub.broadcast(Evo.PubSub, "fund:" <> document.fu
  nd_id, {:record_ready, record.id})
      :ok
    end
  end
end

def handle_event("approve", %{"id" => id}, socket) do
  {:ok, diff} = Review.approve(id, by: socket.assigns.current_use
  r)
  {:noreply, stream_insert(socket, :diffs, diff)}
end

def handle_event("reject", %{"id" => id, "reason" => reason}, soc
  ket) do
  {:ok, diff} = Review.reject(id, reason: reason)
  {:noreply, stream_delete(socket, :diffs, diff)}
end

defp matches?(lot, nil), do: false

  defp put_closing_balance(changeset) do
    opening = get_field(changeset, :opening_balance)
    pnl = get_field(changeset, :allocated_pnl) || Decimal.new(0)
    contrib = get_field(changeset, :contributions)
    dist = get_field(changeset, :distributions)

    closing =
      opening
      |> Decimal.add(contrib)
      |> Decimal.sub(dist)
      |> Decimal.add(pnl)

    put_change(changeset, :closing_balance, closing)
  end
end

defmodule Evo.Workers.ExtractStatement do
  use Oban.Worker, queue: :documents, max_attempts: 5

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"document_id" => id}}) do
    document = Documents.get!(id)

    with {:ok, pages} <- Documents.pages(document),
         {:ok, fields} <- Extraction.run(pages, schema: :custodia
  n_statement),
         {:ok, record} <- Records.upsert(document, fields) do
      Lineage.attach(record, document, fields.sources)
      Phoenix.PubSub.broadcast(Evo.PubSub, "fund:" <> document.fu
  nd_id, {:record_ready, record.id})
      :ok
    end
  end
end

def handle_event("approve", %{"id" => id}, socket) do
  {:ok, diff} = Review.approve(id, by: socket.assigns.current_use
  r)
  {:noreply, stream_insert(socket, :diffs, diff)}
end

def handle_event("reject", %{"id" => id, "reason" => reason}, soc
  ket) do
  {:ok, diff} = Review.reject(id, reason: reason)
  {:noreply, stream_delete(socket, :diffs, diff)}
end

defp matches?(lot, nil), do: false