Add bond full milestone: profile probe, coupon/accrual rules, ledger and point-in-time valuation (3d-29)

1 files changed, 444 insertions, 10 deletions

diff --git a/src/FundLab.Api/Persistence.fs b/src/FundLab.Api/Persistence.fs
index 1cf68cd..2471c58 100644
--- a/src/FundLab.Api/Persistence.fs
+++ b/src/FundLab.Api/Persistence.fs
@@ -363,7 +363,18 @@ type BondTradeCommand =
         InstrumentCode: string
         BondName: string option
         Quantity: decimal
+        /// All-in (dirty/全价) execution price per 100 of face value.
         Price: decimal
+        CleanPrice: decimal
+        AccruedInterest: decimal
+        ParValue: decimal
+        SettlementDate: DateOnly
+        CouponRate: decimal option
+        ValueDate: DateOnly option
+        MaturityDate: DateOnly option
+        /// Explicit trade/valuation date supplied by the caller (wins over the
+        /// quote's own date so tests and backfills stay deterministic).
+        TradeDate: DateOnly option
     }
 
 type BondTradeRecord =
@@ -374,11 +385,55 @@ type BondTradeRecord =
         BondName: string option
         Quantity: decimal
         Price: decimal
+        CleanPrice: decimal
+        AccruedInterest: decimal
+        ParValue: decimal
+        SettlementDate: DateOnly
+        CouponRate: decimal option
+        ValueDate: DateOnly option
+        MaturityDate: DateOnly option
+        TradeDate: DateOnly
         CostCash: decimal
         IsSynthetic: bool
         ExecutedAt: DateTimeOffset
     }
 
+type BondCashflowCommand =
+    {
+        InstrumentCode: string
+        BondName: string option
+        /// "coupon" (付息) or "maturity" (到期).
+        EventType: string
+        EventDate: DateOnly
+        Quantity: decimal
+        /// Cash credited to the fund's available cash.
+        Amount: decimal
+        Note: string option
+    }
+
+type BondCashflowRecord =
+    {
+        Id: Guid
+        FundId: Guid
+        InstrumentCode: string
+        BondName: string option
+        EventType: string
+        EventDate: DateOnly
+        Quantity: decimal
+        Amount: decimal
+        Note: string option
+        IsSynthetic: bool
+        CreatedAt: DateTimeOffset
+    }
+
+type BondCashflowWriteResult =
+    | BondCashflowCreated of BondCashflowRecord
+    | BondCashflowReplayed of BondCashflowRecord
+    | BondCashflowIdempotencyConflict
+    | BondCashflowInvalid of string
+    | BondCashflowFundNotFound
+    | BondCashflowPositionNotFound
+
 type BondPositionRecord =
     {
         FundId: Guid
@@ -1019,6 +1074,15 @@ type FundRepository(connectionString: string) =
             executed_at timestamptz NOT NULL
         );
 
+        ALTER TABLE bond_trades ADD COLUMN IF NOT EXISTS clean_price numeric(20, 4) NOT NULL DEFAULT 0;
+        ALTER TABLE bond_trades ADD COLUMN IF NOT EXISTS accrued_interest numeric(20, 4) NOT NULL DEFAULT 0;
+        ALTER TABLE bond_trades ADD COLUMN IF NOT EXISTS par_value numeric(20, 4) NOT NULL DEFAULT 100;
+        ALTER TABLE bond_trades ADD COLUMN IF NOT EXISTS settlement_date date NOT NULL DEFAULT CURRENT_DATE;
+        ALTER TABLE bond_trades ADD COLUMN IF NOT EXISTS coupon_rate numeric(12, 6) NULL;
+        ALTER TABLE bond_trades ADD COLUMN IF NOT EXISTS value_date date NULL;
+        ALTER TABLE bond_trades ADD COLUMN IF NOT EXISTS maturity_date date NULL;
+        ALTER TABLE bond_trades ADD COLUMN IF NOT EXISTS trade_date date NOT NULL DEFAULT CURRENT_DATE;
+
         CREATE TABLE IF NOT EXISTS bond_trade_idempotencies (
             idempotency_key text PRIMARY KEY,
             request_hash text NOT NULL,
@@ -1027,6 +1091,28 @@ type FundRepository(connectionString: string) =
             created_at timestamptz NOT NULL DEFAULT now()
         );
 
+        CREATE TABLE IF NOT EXISTS bond_cashflow_events (
+            id uuid PRIMARY KEY,
+            fund_id uuid NOT NULL REFERENCES funds(id),
+            instrument_code text NOT NULL,
+            bond_name text NULL,
+            event_type text NOT NULL CHECK (event_type IN ('coupon', 'maturity')),
+            event_date date NOT NULL,
+            quantity numeric(28, 8) NOT NULL CHECK (quantity > 0),
+            amount numeric(20, 2) NOT NULL CHECK (amount >= 0),
+            note text NULL,
+            is_synthetic boolean NOT NULL,
+            created_at timestamptz NOT NULL
+        );
+
+        CREATE TABLE IF NOT EXISTS bond_cashflow_idempotencies (
+            idempotency_key text PRIMARY KEY,
+            request_hash text NOT NULL,
+            event_id uuid NOT NULL REFERENCES bond_cashflow_events(id),
+            fund_id uuid NOT NULL REFERENCES funds(id),
+            created_at timestamptz NOT NULL DEFAULT now()
+        );
+
         CREATE TABLE IF NOT EXISTS bond_positions (
             fund_id uuid NOT NULL REFERENCES funds(id),
             instrument_code text NOT NULL,
@@ -2157,11 +2243,22 @@ type FundRepository(connectionString: string) =
             BondName = if reader.IsDBNull(3) then None else Some(reader.GetString(3))
             Quantity = reader.GetDecimal(4)
             Price = reader.GetDecimal(5)
-            CostCash = reader.GetDecimal(6)
-            IsSynthetic = reader.GetBoolean(7)
-            ExecutedAt = reader.GetFieldValue<DateTimeOffset>(8)
+            CleanPrice = reader.GetDecimal(6)
+            AccruedInterest = reader.GetDecimal(7)
+            ParValue = reader.GetDecimal(8)
+            SettlementDate = reader.GetFieldValue<DateOnly>(9)
+            CouponRate = readDecimalOption reader 10
+            ValueDate = if reader.IsDBNull(11) then None else Some(reader.GetFieldValue<DateOnly>(11))
+            MaturityDate = if reader.IsDBNull(12) then None else Some(reader.GetFieldValue<DateOnly>(12))
+            TradeDate = reader.GetFieldValue<DateOnly>(13)
+            CostCash = reader.GetDecimal(14)
+            IsSynthetic = reader.GetBoolean(15)
+            ExecutedAt = reader.GetFieldValue<DateTimeOffset>(16)
         }
 
+    let bondTradeColumns =
+        "id, fund_id, instrument_code, bond_name, quantity, price, clean_price, accrued_interest, par_value, settlement_date, coupon_rate, value_date, maturity_date, trade_date, cost_cash, is_synthetic, executed_at"
+
     let insertBondTrade connection transaction (trade: BondTradeRecord) =
         use command =
             commandWithTransaction
@@ -2169,9 +2266,13 @@ type FundRepository(connectionString: string) =
                 transaction
                 """
                 INSERT INTO bond_trades
-                    (id, fund_id, instrument_code, bond_name, quantity, price, cost_cash, is_synthetic, executed_at)
+                    (id, fund_id, instrument_code, bond_name, quantity, price, clean_price, accrued_interest,
+                     par_value, settlement_date, coupon_rate, value_date, maturity_date, trade_date, cost_cash,
+                     is_synthetic, executed_at)
                 VALUES
-                    (@id, @fund_id, @instrument_code, @bond_name, @quantity, @price, @cost_cash, @is_synthetic, @executed_at)
+                    (@id, @fund_id, @instrument_code, @bond_name, @quantity, @price, @clean_price, @accrued_interest,
+                     @par_value, @settlement_date, @coupon_rate, @value_date, @maturity_date, @trade_date, @cost_cash,
+                     @is_synthetic, @executed_at)
                 """
 
         addParameter command "id" NpgsqlDbType.Uuid (box trade.Id) |> ignore
@@ -2186,6 +2287,32 @@ type FundRepository(connectionString: string) =
         addParameter command "bond_name" NpgsqlDbType.Text nameParameter |> ignore
         addParameter command "quantity" NpgsqlDbType.Numeric (box trade.Quantity) |> ignore
         addParameter command "price" NpgsqlDbType.Numeric (box trade.Price) |> ignore
+        addParameter command "clean_price" NpgsqlDbType.Numeric (box trade.CleanPrice) |> ignore
+        addParameter command "accrued_interest" NpgsqlDbType.Numeric (box trade.AccruedInterest) |> ignore
+        addParameter command "par_value" NpgsqlDbType.Numeric (box trade.ParValue) |> ignore
+        addParameter command "settlement_date" NpgsqlDbType.Date (box trade.SettlementDate) |> ignore
+
+        let couponParameter =
+            match trade.CouponRate with
+            | Some value -> box value
+            | None -> box DBNull.Value
+
+        addParameter command "coupon_rate" NpgsqlDbType.Numeric couponParameter |> ignore
+
+        let valueDateParameter =
+            match trade.ValueDate with
+            | Some value -> box value
+            | None -> box DBNull.Value
+
+        addParameter command "value_date" NpgsqlDbType.Date valueDateParameter |> ignore
+
+        let maturityParameter =
+            match trade.MaturityDate with
+            | Some value -> box value
+            | None -> box DBNull.Value
+
+        addParameter command "maturity_date" NpgsqlDbType.Date maturityParameter |> ignore
+        addParameter command "trade_date" NpgsqlDbType.Date (box trade.TradeDate) |> ignore
         addParameter command "cost_cash" NpgsqlDbType.Numeric (box trade.CostCash) |> ignore
         addParameter command "is_synthetic" NpgsqlDbType.Boolean (box trade.IsSynthetic) |> ignore
         addParameter command "executed_at" NpgsqlDbType.TimestampTz (box trade.ExecutedAt) |> ignore
@@ -2232,8 +2359,8 @@ type FundRepository(connectionString: string) =
             commandWithTransaction
                 connection
                 transaction
-                """
-                SELECT id, fund_id, instrument_code, bond_name, quantity, price, cost_cash, is_synthetic, executed_at
+                $"""
+                SELECT {bondTradeColumns}
                 FROM bond_trades
                 WHERE id = @id
                 """
@@ -2251,6 +2378,11 @@ type FundRepository(connectionString: string) =
         let invariant = CultureInfo.InvariantCulture
         let encoded (value: string) = sprintf "%d:%s" value.Length value
         let name = command.BondName |> Option.defaultValue ""
+        let dateText (value: DateOnly) = value.ToString("yyyy-MM-dd", invariant)
+        let optionDate = command.ValueDate |> Option.map dateText |> Option.defaultValue ""
+        let optionMaturity = command.MaturityDate |> Option.map dateText |> Option.defaultValue ""
+        let optionCoupon = command.CouponRate |> Option.map (fun v -> v.ToString("G29", invariant)) |> Option.defaultValue ""
+        let optionTradeDate = command.TradeDate |> Option.map dateText |> Option.defaultValue ""
 
         let payload =
             String.concat
@@ -2262,6 +2394,148 @@ type FundRepository(connectionString: string) =
                     encoded name
                     encoded (command.Quantity.ToString("G29", invariant))
                     encoded (command.Price.ToString("G29", invariant))
+                    encoded (command.CleanPrice.ToString("G29", invariant))
+                    encoded (command.AccruedInterest.ToString("G29", invariant))
+                    encoded (command.ParValue.ToString("G29", invariant))
+                    encoded (dateText command.SettlementDate)
+                    encoded optionTradeDate
+                    encoded optionCoupon
+                    encoded optionDate
+                    encoded optionMaturity
+                ]
+
+        Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(payload)))
+
+    let bondCashflowRecordFromReader (reader: DbDataReader) : BondCashflowRecord =
+        {
+            Id = reader.GetGuid(0)
+            FundId = reader.GetGuid(1)
+            InstrumentCode = reader.GetString(2)
+            BondName = if reader.IsDBNull(3) then None else Some(reader.GetString(3))
+            EventType = reader.GetString(4)
+            EventDate = reader.GetFieldValue<DateOnly>(5)
+            Quantity = reader.GetDecimal(6)
+            Amount = reader.GetDecimal(7)
+            Note = readStringOption reader 8
+            IsSynthetic = reader.GetBoolean(9)
+            CreatedAt = reader.GetFieldValue<DateTimeOffset>(10)
+        }
+
+    let bondCashflowColumns =
+        "id, fund_id, instrument_code, bond_name, event_type, event_date, quantity, amount, note, is_synthetic, created_at"
+
+    let insertBondCashflow connection transaction (record: BondCashflowRecord) =
+        use command =
+            commandWithTransaction
+                connection
+                transaction
+                """
+                INSERT INTO bond_cashflow_events
+                    (id, fund_id, instrument_code, bond_name, event_type, event_date, quantity, amount, note, is_synthetic, created_at)
+                VALUES
+                    (@id, @fund_id, @instrument_code, @bond_name, @event_type, @event_date, @quantity, @amount, @note, @is_synthetic, @created_at)
+                """
+
+        addParameter command "id" NpgsqlDbType.Uuid (box record.Id) |> ignore
+        addParameter command "fund_id" NpgsqlDbType.Uuid (box record.FundId) |> ignore
+        addParameter command "instrument_code" NpgsqlDbType.Text (box record.InstrumentCode) |> ignore
+
+        let nameParameter =
+            match record.BondName with
+            | Some name -> box name
+            | None -> box DBNull.Value
+
+        addParameter command "bond_name" NpgsqlDbType.Text nameParameter |> ignore
+        addParameter command "event_type" NpgsqlDbType.Text (box record.EventType) |> ignore
+        addParameter command "event_date" NpgsqlDbType.Date (box record.EventDate) |> ignore
+        addParameter command "quantity" NpgsqlDbType.Numeric (box record.Quantity) |> ignore
+        addParameter command "amount" NpgsqlDbType.Numeric (box record.Amount) |> ignore
+
+        let noteParameter =
+            match record.Note with
+            | Some note -> box note
+            | None -> box DBNull.Value
+
+        addParameter command "note" NpgsqlDbType.Text noteParameter |> ignore
+        addParameter command "is_synthetic" NpgsqlDbType.Boolean (box record.IsSynthetic) |> ignore
+        addParameter command "created_at" NpgsqlDbType.TimestampTz (box record.CreatedAt) |> ignore
+        command.ExecuteNonQuery() |> ignore
+
+    let insertBondCashflowIdempotency connection transaction key requestHash eventId fundId =
+        use command =
+            commandWithTransaction
+                connection
+                transaction
+                """
+                INSERT INTO bond_cashflow_idempotencies (idempotency_key, request_hash, event_id, fund_id)
+                VALUES (@idempotency_key, @request_hash, @event_id, @fund_id)
+                """
+
+        addParameter command "idempotency_key" NpgsqlDbType.Text (box key) |> ignore
+        addParameter command "request_hash" NpgsqlDbType.Text (box requestHash) |> ignore
+        addParameter command "event_id" NpgsqlDbType.Uuid (box eventId) |> ignore
+        addParameter command "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore
+        command.ExecuteNonQuery() |> ignore
+
+    let findBondCashflowIdempotency connection transaction key =
+        use command =
+            commandWithTransaction
+                connection
+                transaction
+                """
+                SELECT request_hash, fund_id, event_id
+                FROM bond_cashflow_idempotencies
+                WHERE idempotency_key = @idempotency_key
+                """
+
+        addParameter command "idempotency_key" NpgsqlDbType.Text (box key) |> ignore
+
+        use reader = command.ExecuteReader()
+
+        if reader.Read() then
+            Some(reader.GetString(0), reader.GetGuid(1), reader.GetGuid(2))
+        else
+            None
+
+    let findBondCashflow connection transaction eventId =
+        use command =
+            commandWithTransaction
+                connection
+                transaction
+                $"""
+                SELECT {bondCashflowColumns}
+                FROM bond_cashflow_events
+                WHERE id = @id
+                """
+
+        addParameter command "id" NpgsqlDbType.Uuid (box eventId) |> ignore
+
+        use reader = command.ExecuteReader()
+
+        if reader.Read() then
+            Some(bondCashflowRecordFromReader reader)
+        else
+            None
+
+    let bondCashflowRequestHash (fundId: Guid) (command: BondCashflowCommand) =
+        let invariant = CultureInfo.InvariantCulture
+        let encoded (value: string) = sprintf "%d:%s" value.Length value
+        let name = command.BondName |> Option.defaultValue ""
+        let note = command.Note |> Option.defaultValue ""
+
+        let payload =
+            String.concat
+                "|"
+                [
+                    "bond-cashflow"
+                    encoded (fundId.ToString("D"))
+                    encoded command.InstrumentCode
+                    encoded name
+                    encoded command.EventType
+                    encoded (command.EventDate.ToString("yyyy-MM-dd", invariant))
+                    encoded (command.Quantity.ToString("G29", invariant))
+                    encoded (command.Amount.ToString("G29", invariant))
+                    encoded note
                 ]
 
         Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(payload)))
@@ -5169,6 +5443,12 @@ type FundRepository(connectionString: string) =
                 BondTradeWriteResult.BondTradeInvalid "quantity must be positive"
             elif command.Price <= 0m then
                 BondTradeWriteResult.BondTradeInvalid "price must be positive"
+            elif command.CleanPrice <= 0m then
+                BondTradeWriteResult.BondTradeInvalid "clean price must be positive"
+            elif command.ParValue <= 0m then
+                BondTradeWriteResult.BondTradeInvalid "par value must be positive"
+            elif command.AccruedInterest < 0m then
+                BondTradeWriteResult.BondTradeInvalid "accrued interest cannot be negative"
             else
                 let normalized = { command with InstrumentCode = code }
                 let fingerprint = bondTradeRequestHash fundId normalized
@@ -5206,7 +5486,13 @@ type FundRepository(connectionString: string) =
                             BondTradeWriteResult.BondTradeFundNotFound
                         | Some isSynthetic ->
                             let executedAt = defaultArg executedAtOverride DateTimeOffset.UtcNow
-                            let costCash = Decimal.Round(normalized.Quantity * normalized.Price, 2, MidpointRounding.AwayFromZero)
+
+                            let costCash =
+                                Decimal.Round(
+                                    normalized.Quantity * normalized.Price * normalized.ParValue / 100m,
+                                    2,
+                                    MidpointRounding.AwayFromZero
+                                )
 
                             let trade: BondTradeRecord =
                                 {
@@ -5216,6 +5502,14 @@ type FundRepository(connectionString: string) =
                                     BondName = normalized.BondName
                                     Quantity = normalized.Quantity
                                     Price = normalized.Price
+                                    CleanPrice = normalized.CleanPrice
+                                    AccruedInterest = normalized.AccruedInterest
+                                    ParValue = normalized.ParValue
+                                    SettlementDate = normalized.SettlementDate
+                                    CouponRate = normalized.CouponRate
+                                    ValueDate = normalized.ValueDate
+                                    MaturityDate = normalized.MaturityDate
+                                    TradeDate = normalized.TradeDate |> Option.defaultValue normalized.SettlementDate
                                     CostCash = costCash
                                     IsSynthetic = isSynthetic
                                     ExecutedAt = executedAt
@@ -5271,8 +5565,8 @@ type FundRepository(connectionString: string) =
             commandWithTransaction
                 connection
                 None
-                """
-                SELECT id, fund_id, instrument_code, bond_name, quantity, price, cost_cash, is_synthetic, executed_at
+                $"""
+                SELECT {bondTradeColumns}
                 FROM bond_trades
                 WHERE fund_id = @fund_id
                 ORDER BY executed_at, id
@@ -5288,6 +5582,146 @@ type FundRepository(connectionString: string) =
 
         records |> Seq.toList
 
+    member _.RecordBondCashflow(idempotencyKey: string, fundId: Guid, command: BondCashflowCommand) : BondCashflowWriteResult =
+        let code = if isNull command.InstrumentCode then "" else command.InstrumentCode.Trim()
+        let eventType = if isNull command.EventType then "" else command.EventType.Trim().ToLowerInvariant()
+
+        if String.IsNullOrWhiteSpace idempotencyKey then
+            BondCashflowWriteResult.BondCashflowInvalid "idempotency key cannot be empty"
+        elif code.Length <> 6 || not (code |> Seq.forall Char.IsDigit) then
+            BondCashflowWriteResult.BondCashflowInvalid "bond code must contain exactly six digits"
+        elif eventType <> "coupon" && eventType <> "maturity" then
+            BondCashflowWriteResult.BondCashflowInvalid "event type must be coupon or maturity"
+        elif command.Quantity <= 0m then
+            BondCashflowWriteResult.BondCashflowInvalid "quantity must be positive"
+        elif command.Amount < 0m then
+            BondCashflowWriteResult.BondCashflowInvalid "amount cannot be negative"
+        else
+            let normalized = { command with InstrumentCode = code; EventType = eventType }
+            let fingerprint = bondCashflowRequestHash fundId normalized
+            use connection = new NpgsqlConnection(connectionString)
+            connection.Open()
+            use transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted)
+
+            try
+                use lockCommand =
+                    commandWithTransaction
+                        connection
+                        (Some transaction)
+                        "SELECT pg_advisory_xact_lock(hashtext(@lock_key))"
+
+                addParameter lockCommand "lock_key" NpgsqlDbType.Text (box idempotencyKey) |> ignore
+                lockCommand.ExecuteNonQuery() |> ignore
+
+                match findBondCashflowIdempotency connection (Some transaction) idempotencyKey with
+                | Some(existingHash, existingFundId, eventId)
+                    when existingHash = fingerprint && existingFundId = fundId ->
+                    match findBondCashflow connection (Some transaction) eventId with
+                    | Some record ->
+                        transaction.Commit()
+                        BondCashflowWriteResult.BondCashflowReplayed record
+                    | None ->
+                        transaction.Rollback()
+                        BondCashflowWriteResult.BondCashflowInvalid "idempotency record references a missing event"
+                | Some _ ->
+                    transaction.Rollback()
+                    BondCashflowWriteResult.BondCashflowIdempotencyConflict
+                | None ->
+                    match lockFundForOrder connection (Some transaction) fundId with
+                    | None ->
+                        transaction.Rollback()
+                        BondCashflowWriteResult.BondCashflowFundNotFound
+                    | Some isSynthetic ->
+                        let hasPosition =
+                            use positionQuery =
+                                commandWithTransaction
+                                    connection
+                                    (Some transaction)
+                                    "SELECT 1 FROM bond_positions WHERE fund_id = @fund_id AND instrument_code = @code"
+
+                            addParameter positionQuery "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore
+                            addParameter positionQuery "code" NpgsqlDbType.Text (box normalized.InstrumentCode) |> ignore
+                            use reader = positionQuery.ExecuteReader()
+                            reader.Read()
+
+                        if not hasPosition then
+                            transaction.Rollback()
+                            BondCashflowWriteResult.BondCashflowPositionNotFound
+                        else
+                            let record: BondCashflowRecord =
+                                {
+                                    Id = Guid.NewGuid()
+                                    FundId = fundId
+                                    InstrumentCode = normalized.InstrumentCode
+                                    BondName = normalized.BondName
+                                    EventType = normalized.EventType
+                                    EventDate = normalized.EventDate
+                                    Quantity = normalized.Quantity
+                                    Amount = normalized.Amount
+                                    Note = normalized.Note
+                                    IsSynthetic = isSynthetic
+                                    CreatedAt = DateTimeOffset.UtcNow
+                                }
+
+                            insertBondCashflow connection (Some transaction) record
+                            insertBondCashflowIdempotency connection (Some transaction) idempotencyKey fingerprint record.Id fundId
+
+                            use cashCommand =
+                                commandWithTransaction
+                                    connection
+                                    (Some transaction)
+                                    "UPDATE funds SET available_cash = available_cash + @amount WHERE id = @fund_id"
+
+                            addParameter cashCommand "amount" NpgsqlDbType.Numeric (box normalized.Amount) |> ignore
+                            addParameter cashCommand "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore
+                            cashCommand.ExecuteNonQuery() |> ignore
+
+                            if normalized.EventType = "maturity" then
+                                use removeCommand =
+                                    commandWithTransaction
+                                        connection
+                                        (Some transaction)
+                                        "DELETE FROM bond_positions WHERE fund_id = @fund_id AND instrument_code = @code"
+
+                                addParameter removeCommand "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore
+                                addParameter removeCommand "code" NpgsqlDbType.Text (box normalized.InstrumentCode) |> ignore
+                                removeCommand.ExecuteNonQuery() |> ignore
+
+                            transaction.Commit()
+                            BondCashflowWriteResult.BondCashflowCreated record
+            with error ->
+                try
+                    transaction.Rollback()
+                with _ ->
+                    ()
+
+                raise error
+
+    member _.GetBondCashflows(fundId: Guid) : BondCashflowRecord list =
+        use connection = new NpgsqlConnection(connectionString)
+        connection.Open()
+
+        use command =
+            commandWithTransaction
+                connection
+                None
+                $"""
+                SELECT {bondCashflowColumns}
+                FROM bond_cashflow_events
+                WHERE fund_id = @fund_id
+                ORDER BY event_date, created_at, id
+                """
+
+        addParameter command "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore
+
+        use reader = command.ExecuteReader()
+        let records = ResizeArray<BondCashflowRecord>()
+
+        while reader.Read() do
+            records.Add(bondCashflowRecordFromReader reader)
+
+        records |> Seq.toList
+
     member _.GetBondPositions(fundId: Guid) : BondPositionRecord list =
         use connection = new NpgsqlConnection(connectionString)
         connection.Open()