Blizzard
92329ddce2
Checks / verify (push) Failing after 9m52s
统一单入口 Go 工程与前端依赖目录，增加后台供应商配置、候选审核、计划预算、成交冲正、持仓复盘和站内通知。接入策略版本、Python 回测、可配置 A 股公开行情及 zap 文件轮转日志，补充迁移、契约与验证。
335 lines
16 KiB
Go
335 lines
16 KiB
Go
package repository
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/model"
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
"strconv"
|
|
"time"
|
|
)
|
|
|
|
type TradeFillRepository struct{ pool *pgxpool.Pool }
|
|
|
|
func NewTradeFillRepository(pool *pgxpool.Pool) *TradeFillRepository {
|
|
return &TradeFillRepository{pool}
|
|
}
|
|
|
|
const fillColumns = `f.id,f.account_id,f.plan_id,f.idempotency_key,f.input_hash,f.posted_sequence,f.side,f.symbol,f.exchange,f.name,f.quantity,f.price_minor,f.fee_minor,f.cash_delta_minor,f.realized_minor,f.occurred_at,f.sellable_at,f.note,f.created_at,EXISTS(SELECT 1 FROM veyra_trade_reversal x WHERE x.fill_id=f.id)`
|
|
|
|
func scanFill(row pgx.Row) (model.TradeFill, error) {
|
|
var f model.TradeFill
|
|
err := row.Scan(&f.ID, &f.AccountID, &f.PlanID, &f.IdempotencyKey, &f.InputHash, &f.PostedSequence, &f.Side, &f.Symbol, &f.Exchange, &f.Name, &f.Quantity, &f.PriceMinor, &f.FeeMinor, &f.CashDeltaMinor, &f.RealizedMinor, &f.OccurredAt, &f.SellableAt, &f.Note, &f.CreatedAt, &f.Reversed)
|
|
f.Price = model.FormatMoney(f.PriceMinor)
|
|
f.Fee = model.FormatMoney(f.FeeMinor)
|
|
f.CashDelta = model.FormatMoney(f.CashDeltaMinor)
|
|
f.RealizedPnL = model.FormatMoney(f.RealizedMinor)
|
|
return f, mapError(err)
|
|
}
|
|
func appendPositionEvent(ctx context.Context, tx pgx.Tx, account, fill, aggregateType, eventType string) error {
|
|
payload, err := json.Marshal(map[string]string{"account_id": account, "fill_id": fill})
|
|
if err != nil {
|
|
return apperror.ErrUnavailable
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO veyra_outbox_event(id,aggregate_type,aggregate_id,aggregate_version,event_type,payload) VALUES($1,$2,$3,1,$4,$5)`, uuid.NewString(), aggregateType, fill, eventType, payload)
|
|
return mapError(err)
|
|
}
|
|
func (r *TradeFillRepository) Record(ctx context.Context, user string, fill model.TradeFill, requestID string) (model.TradeFill, error) {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return fill, mapError(err)
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
if err := lockSimulationAccount(ctx, tx, user, fill.AccountID); err != nil {
|
|
return fill, err
|
|
}
|
|
previous, err := scanFill(tx.QueryRow(ctx, `SELECT `+fillColumns+` FROM veyra_trade_fill f WHERE account_id=$1 AND idempotency_key=$2`, fill.AccountID, fill.IdempotencyKey))
|
|
if err == nil {
|
|
if previous.InputHash != fill.InputHash {
|
|
return fill, apperror.ErrConflict
|
|
}
|
|
return previous, nil
|
|
}
|
|
if err != apperror.ErrNotFound {
|
|
return fill, err
|
|
}
|
|
var latest *time.Time
|
|
if err := tx.QueryRow(ctx, `SELECT max(f.occurred_at) FROM veyra_trade_fill f WHERE f.account_id=$1 AND NOT EXISTS(SELECT 1 FROM veyra_trade_reversal x WHERE x.fill_id=f.id)`, fill.AccountID).Scan(&latest); err != nil {
|
|
return fill, mapError(err)
|
|
}
|
|
if latest != nil && fill.OccurredAt.Before(*latest) {
|
|
return fill, apperror.ErrChronology
|
|
}
|
|
var plan *model.TradePlan
|
|
exclude := ""
|
|
if fill.PlanID != nil {
|
|
p, err := scanTradePlan(tx.QueryRow(ctx, `SELECT `+tradePlanColumns+tradePlanJoin+` WHERE p.id=$1 AND p.account_id=$2 FOR UPDATE OF p`, *fill.PlanID, fill.AccountID))
|
|
if err != nil {
|
|
return fill, err
|
|
}
|
|
if p.CandidateContent.Symbol != fill.Symbol || p.CandidateContent.Exchange != fill.Exchange || fill.Quantity > p.Quantity-p.FilledQuantity {
|
|
return fill, apperror.ErrInvalidInput
|
|
}
|
|
low, e1 := model.ParseMoney(p.CandidateContent.ReferencePlan.EntryLow)
|
|
high, e2 := model.ParseMoney(p.CandidateContent.ReferencePlan.EntryHigh)
|
|
if e1 != nil || e2 != nil {
|
|
return fill, apperror.ErrUnavailable
|
|
}
|
|
if (fill.PriceMinor < low || fill.PriceMinor > high || p.Status != "ACTIVE" && p.Status != "PART_FILLED") && fill.Note == "" {
|
|
return fill, apperror.ErrInvalidInput
|
|
}
|
|
plan = &p
|
|
exclude = p.ID
|
|
}
|
|
cash, reserved, err := budgetMinor(ctx, tx, fill.AccountID, exclude)
|
|
if err != nil {
|
|
return fill, err
|
|
}
|
|
if fill.Side == "BUY" && cash+fill.CashDeltaMinor < reserved {
|
|
return fill, apperror.ErrInsufficientCash
|
|
}
|
|
allocations := make([]model.FillAllocation, 0)
|
|
if fill.Side == "SELL" {
|
|
rows, err := tx.Query(ctx, `SELECT l.id,l.remaining_quantity,l.remaining_cost_minor FROM veyra_position_lot l JOIN veyra_trade_fill f ON f.id=l.buy_fill_id WHERE l.account_id=$1 AND l.symbol=$2 AND l.exchange=$3 AND l.remaining_quantity>0 AND l.sellable_at<=$4 ORDER BY f.occurred_at,f.posted_sequence FOR UPDATE OF l`, fill.AccountID, fill.Symbol, fill.Exchange, fill.OccurredAt)
|
|
if err != nil {
|
|
return fill, mapError(err)
|
|
}
|
|
remaining, cost := fill.Quantity, int64(0)
|
|
for rows.Next() {
|
|
var id string
|
|
var qty, minor int64
|
|
if err := rows.Scan(&id, &qty, &minor); err != nil {
|
|
rows.Close()
|
|
return fill, mapError(err)
|
|
}
|
|
if remaining == 0 {
|
|
continue
|
|
}
|
|
take := qty
|
|
if remaining < take {
|
|
take = remaining
|
|
}
|
|
allocated := minor * take / qty
|
|
allocations = append(allocations, model.FillAllocation{ID: uuid.NewString(), SellFillID: fill.ID, LotID: id, Quantity: take, CostMinor: allocated})
|
|
remaining -= take
|
|
cost += allocated
|
|
}
|
|
rows.Close()
|
|
if err := rows.Err(); err != nil {
|
|
return fill, mapError(err)
|
|
}
|
|
if remaining != 0 {
|
|
return fill, apperror.ErrInsufficientQuantity
|
|
}
|
|
fill.RealizedMinor = fill.CashDeltaMinor - cost
|
|
}
|
|
err = tx.QueryRow(ctx, `INSERT INTO veyra_trade_fill(id,account_id,plan_id,idempotency_key,input_hash,side,symbol,exchange,name,quantity,price_minor,fee_minor,cash_delta_minor,realized_minor,occurred_at,sellable_at,note) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17) RETURNING posted_sequence,created_at`, fill.ID, fill.AccountID, fill.PlanID, fill.IdempotencyKey, fill.InputHash, fill.Side, fill.Symbol, fill.Exchange, fill.Name, fill.Quantity, fill.PriceMinor, fill.FeeMinor, fill.CashDeltaMinor, fill.RealizedMinor, fill.OccurredAt, fill.SellableAt, fill.Note).Scan(&fill.PostedSequence, &fill.CreatedAt)
|
|
if err != nil {
|
|
return fill, mapError(err)
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO veyra_cash_entry(id,account_id,entry_type,amount_minor,fill_id) VALUES($1,$2,$3,$4,$5)`, uuid.NewString(), fill.AccountID, fill.Side, fill.CashDeltaMinor, fill.ID)
|
|
if err != nil {
|
|
return fill, mapError(err)
|
|
}
|
|
if fill.Side == "BUY" {
|
|
_, err = tx.Exec(ctx, `INSERT INTO veyra_position_lot(id,account_id,buy_fill_id,symbol,exchange,remaining_quantity,remaining_cost_minor,sellable_at) VALUES($1,$2,$3,$4,$5,$6,$7,$8)`, uuid.NewString(), fill.AccountID, fill.ID, fill.Symbol, fill.Exchange, fill.Quantity, -fill.CashDeltaMinor, fill.SellableAt)
|
|
if err != nil {
|
|
return fill, mapError(err)
|
|
}
|
|
} else {
|
|
for _, a := range allocations {
|
|
_, err = tx.Exec(ctx, `INSERT INTO veyra_fill_allocation(id,sell_fill_id,lot_id,quantity,cost_minor) VALUES($1,$2,$3,$4,$5)`, a.ID, a.SellFillID, a.LotID, a.Quantity, a.CostMinor)
|
|
if err != nil {
|
|
return fill, mapError(err)
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE veyra_position_lot SET remaining_quantity=remaining_quantity-$2,remaining_cost_minor=remaining_cost_minor-$3 WHERE id=$1`, a.LotID, a.Quantity, a.CostMinor)
|
|
if err != nil {
|
|
return fill, mapError(err)
|
|
}
|
|
}
|
|
}
|
|
if plan != nil {
|
|
remaining := plan.Quantity - plan.FilledQuantity - fill.Quantity
|
|
high, err := model.ParseMoney(plan.CandidateContent.ReferencePlan.EntryHigh)
|
|
if err != nil {
|
|
return fill, err
|
|
}
|
|
otherReserved := reserved
|
|
newStatus, reserved := plan.Status, int64(0)
|
|
if plan.Status == "ACTIVE" || plan.Status == "PART_FILLED" {
|
|
if remaining == 0 {
|
|
newStatus = "COMPLETED"
|
|
} else {
|
|
newStatus = "PART_FILLED"
|
|
reserved = high*remaining + plan.FeeBudgetMinor
|
|
if cash+fill.CashDeltaMinor < reserved+otherReserved {
|
|
newStatus = "INVALIDATED"
|
|
reserved = 0
|
|
}
|
|
}
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE veyra_trade_plan SET filled_quantity=filled_quantity+$2,reserved_minor=$3,status=$4,version=version+1,updated_at=now() WHERE id=$1`, plan.ID, fill.Quantity, reserved, newStatus)
|
|
if err != nil {
|
|
return fill, mapError(err)
|
|
}
|
|
}
|
|
if err := appendAudit(ctx, tx, user, "FILL_RECORDED", "trade_fill", fill.ID, requestID, map[string]any{"side": fill.Side, "quantity": fill.Quantity, "plan_id": fill.PlanID}); err != nil {
|
|
return fill, err
|
|
}
|
|
if err := appendPositionEvent(ctx, tx, fill.AccountID, fill.ID, "trade_fill", "position.changed"); err != nil {
|
|
return fill, err
|
|
}
|
|
saved, err := scanFill(tx.QueryRow(ctx, `SELECT `+fillColumns+` FROM veyra_trade_fill f WHERE f.id=$1`, fill.ID))
|
|
if err != nil {
|
|
return saved, err
|
|
}
|
|
return saved, mapError(tx.Commit(ctx))
|
|
}
|
|
func (r *TradeFillRepository) List(ctx context.Context, user, account string) ([]model.TradeFill, error) {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, mapError(err)
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
if err := lockSimulationAccount(ctx, tx, user, account); err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := tx.Query(ctx, `SELECT `+fillColumns+` FROM veyra_trade_fill f WHERE f.account_id=$1 ORDER BY f.posted_sequence DESC LIMIT 100`, account)
|
|
if err != nil {
|
|
return nil, mapError(err)
|
|
}
|
|
defer rows.Close()
|
|
list := make([]model.TradeFill, 0)
|
|
for rows.Next() {
|
|
f, err := scanFill(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
list = append(list, f)
|
|
}
|
|
return list, mapError(rows.Err())
|
|
}
|
|
func (r *TradeFillRepository) Reverse(ctx context.Context, user, id, reason, requestID string) (model.TradeReversal, error) {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return model.TradeReversal{}, mapError(err)
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
var account string
|
|
if err := tx.QueryRow(ctx, `SELECT f.account_id FROM veyra_trade_fill f JOIN veyra_account a ON a.id=f.account_id WHERE f.id=$1 AND a.user_id=$2`, id, user).Scan(&account); err != nil {
|
|
return model.TradeReversal{}, mapError(err)
|
|
}
|
|
if err := lockSimulationAccount(ctx, tx, user, account); err != nil {
|
|
return model.TradeReversal{}, err
|
|
}
|
|
var existing model.TradeReversal
|
|
err = tx.QueryRow(ctx, `SELECT id,fill_id,actor_id,reason,created_at FROM veyra_trade_reversal WHERE fill_id=$1`, id).Scan(&existing.ID, &existing.FillID, &existing.ActorID, &existing.Reason, &existing.CreatedAt)
|
|
if err == nil {
|
|
return existing, nil
|
|
}
|
|
if err != pgx.ErrNoRows {
|
|
return existing, mapError(err)
|
|
}
|
|
fill, err := scanFill(tx.QueryRow(ctx, `SELECT `+fillColumns+` FROM veyra_trade_fill f WHERE f.id=$1`, id))
|
|
if err != nil {
|
|
return existing, err
|
|
}
|
|
var later bool
|
|
err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM veyra_trade_fill f WHERE f.account_id=$1 AND f.posted_sequence>$2 AND NOT EXISTS(SELECT 1 FROM veyra_trade_reversal x WHERE x.fill_id=f.id))`, account, fill.PostedSequence).Scan(&later)
|
|
if err != nil {
|
|
return existing, mapError(err)
|
|
}
|
|
if later {
|
|
return existing, apperror.ErrDependentFill
|
|
}
|
|
cash, reserved, err := budgetMinor(ctx, tx, account, "")
|
|
if err != nil {
|
|
return existing, err
|
|
}
|
|
if cash-fill.CashDeltaMinor < reserved {
|
|
return existing, apperror.ErrInsufficientCash
|
|
}
|
|
if fill.Side == "BUY" {
|
|
tag, err := tx.Exec(ctx, `UPDATE veyra_position_lot SET remaining_quantity=0,remaining_cost_minor=0 WHERE buy_fill_id=$1 AND remaining_quantity=$2 AND remaining_cost_minor=$3`, id, fill.Quantity, -fill.CashDeltaMinor)
|
|
if err != nil {
|
|
return existing, mapError(err)
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return existing, apperror.ErrConflict
|
|
}
|
|
} else {
|
|
_, err = tx.Exec(ctx, `UPDATE veyra_position_lot l SET remaining_quantity=l.remaining_quantity+a.quantity,remaining_cost_minor=l.remaining_cost_minor+a.cost_minor FROM veyra_fill_allocation a WHERE a.lot_id=l.id AND a.sell_fill_id=$1`, id)
|
|
if err != nil {
|
|
return existing, mapError(err)
|
|
}
|
|
}
|
|
reversal := model.TradeReversal{ID: uuid.NewString(), FillID: id, ActorID: user, Reason: reason}
|
|
err = tx.QueryRow(ctx, `INSERT INTO veyra_trade_reversal(id,fill_id,actor_id,reason) VALUES($1,$2,$3,$4) RETURNING created_at`, reversal.ID, id, user, reason).Scan(&reversal.CreatedAt)
|
|
if err != nil {
|
|
return reversal, mapError(err)
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO veyra_cash_entry(id,account_id,entry_type,amount_minor,reversal_id) VALUES($1,$2,'REVERSAL',$3,$4)`, uuid.NewString(), account, -fill.CashDeltaMinor, reversal.ID)
|
|
if err != nil {
|
|
return reversal, mapError(err)
|
|
}
|
|
if fill.PlanID != nil {
|
|
_, err = tx.Exec(ctx, `UPDATE veyra_trade_plan SET filled_quantity=filled_quantity-$2,status='INVALIDATED',reserved_minor=0,version=version+1,updated_at=now() WHERE id=$1`, *fill.PlanID, fill.Quantity)
|
|
if err != nil {
|
|
return reversal, mapError(err)
|
|
}
|
|
}
|
|
if err := appendAudit(ctx, tx, user, "FILL_REVERSED", "trade_fill", id, requestID, map[string]any{"reason": reason, "reversal_id": reversal.ID, "plan_reopened": false}); err != nil {
|
|
return reversal, err
|
|
}
|
|
if err := appendPositionEvent(ctx, tx, account, reversal.ID, "trade_reversal", "position.changed"); err != nil {
|
|
return reversal, err
|
|
}
|
|
return reversal, mapError(tx.Commit(ctx))
|
|
}
|
|
func (r *TradeFillRepository) Portfolio(ctx context.Context, user, account string) (model.PortfolioSummary, error) {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return model.PortfolioSummary{}, mapError(err)
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
if err := lockSimulationAccount(ctx, tx, user, account); err != nil {
|
|
return model.PortfolioSummary{}, err
|
|
}
|
|
var cash, realized int64
|
|
if err := tx.QueryRow(ctx, `SELECT COALESCE(sum(amount_minor),0)::bigint FROM veyra_cash_entry WHERE account_id=$1`, account).Scan(&cash); err != nil {
|
|
return model.PortfolioSummary{}, mapError(err)
|
|
}
|
|
if err := tx.QueryRow(ctx, `SELECT COALESCE(sum(f.realized_minor),0)::bigint FROM veyra_trade_fill f WHERE account_id=$1 AND NOT EXISTS(SELECT 1 FROM veyra_trade_reversal x WHERE x.fill_id=f.id)`, account).Scan(&realized); err != nil {
|
|
return model.PortfolioSummary{}, mapError(err)
|
|
}
|
|
portfolio := model.PortfolioSummary{AccountID: account, CashBalance: model.FormatMoney(cash), RealizedPnL: model.FormatMoney(realized), Positions: make([]model.PositionSummary, 0)}
|
|
var buy, sell, fees int64
|
|
err = tx.QueryRow(ctx, `SELECT count(*),count(*) FILTER(WHERE f.side='SELL'),count(*) FILTER(WHERE f.side='SELL' AND realized_minor>0),count(*) FILTER(WHERE f.side='SELL' AND realized_minor<0),COALESCE(sum(-cash_delta_minor) FILTER(WHERE f.side='BUY'),0)::bigint,COALESCE(sum(cash_delta_minor) FILTER(WHERE f.side='SELL'),0)::bigint,COALESCE(sum(fee_minor),0)::bigint FROM veyra_trade_fill f WHERE f.account_id=$1 AND NOT EXISTS(SELECT 1 FROM veyra_trade_reversal x WHERE x.fill_id=f.id)`, account).Scan(&portfolio.Review.TradeCount, &portfolio.Review.SellCount, &portfolio.Review.ProfitableSellCount, &portfolio.Review.LosingSellCount, &buy, &sell, &fees)
|
|
if err != nil {
|
|
return portfolio, mapError(err)
|
|
}
|
|
portfolio.Review.BuyOutflow = model.FormatMoney(buy)
|
|
portfolio.Review.SellInflow = model.FormatMoney(sell)
|
|
portfolio.Review.TotalFees = model.FormatMoney(fees)
|
|
rows, err := tx.Query(ctx, `SELECT l.symbol,l.exchange,min(f.name),sum(l.remaining_quantity)::bigint,COALESCE(sum(l.remaining_quantity) FILTER(WHERE l.sellable_at<=now()),0)::bigint,sum(l.remaining_cost_minor)::bigint FROM veyra_position_lot l JOIN veyra_trade_fill f ON f.id=l.buy_fill_id WHERE l.account_id=$1 AND l.remaining_quantity>0 GROUP BY l.symbol,l.exchange ORDER BY l.exchange,l.symbol`, account)
|
|
if err != nil {
|
|
return portfolio, mapError(err)
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var p model.PositionSummary
|
|
var qty, sellable, cost int64
|
|
if err := rows.Scan(&p.Symbol, &p.Exchange, &p.Name, &qty, &sellable, &cost); err != nil {
|
|
return portfolio, mapError(err)
|
|
}
|
|
p.Quantity = strconv.FormatInt(qty, 10)
|
|
p.SellableQuantity = strconv.FormatInt(sellable, 10)
|
|
p.CostBasis = model.FormatMoney(cost)
|
|
portfolio.Positions = append(portfolio.Positions, p)
|
|
}
|
|
return portfolio, mapError(rows.Err())
|
|
}
|