feat: 完成模拟交易闭环、App 与 A 股行情接入

统一单入口 Go 工程与前端依赖目录，增加后台供应商配置、候选审核、计划预算、成交冲正、持仓复盘和站内通知。接入策略版本、Python 回测、可配置 A 股公开行情及 zap 文件轮转日志，补充迁移、契约与验证。

internal/repository/backtest.go

@@ -0,0 +1,100 @@
+package repository
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"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"
+)
+
+type BacktestRepository struct{ pool *pgxpool.Pool }
+
+func NewBacktestRepository(pool *pgxpool.Pool) *BacktestRepository { return &BacktestRepository{pool} }
+
+const backtestColumns = `id,strategy_version_id,status,input,input_hash,result,failure,lease_id,created_at,finished_at`
+
+func scanBacktest(row pgx.Row) (model.BacktestRun, error) {
+	var v model.BacktestRun
+	err := row.Scan(&v.ID, &v.StrategyVersionID, &v.Status, &v.Input, &v.InputHash, &v.Result, &v.Failure, &v.LeaseID, &v.CreatedAt, &v.FinishedAt)
+	return v, mapError(err)
+}
+func (r *BacktestRepository) Detail(ctx context.Context, id string) (model.BacktestRun, error) {
+	return scanBacktest(r.pool.QueryRow(ctx, `SELECT `+backtestColumns+` FROM veyra_backtest_run WHERE id=$1`, id))
+}
+func (r *BacktestRepository) List(ctx context.Context) ([]model.BacktestRun, error) {
+	rows, e := r.pool.Query(ctx, `SELECT `+backtestColumns+` FROM veyra_backtest_run ORDER BY created_at DESC,id DESC LIMIT 100`)
+	if e != nil {
+		return nil, mapError(e)
+	}
+	defer rows.Close()
+	list := make([]model.BacktestRun, 0)
+	for rows.Next() {
+		v, e := scanBacktest(rows)
+		if e != nil {
+			return nil, e
+		}
+		v.Input = nil
+		v.Result = nil
+		list = append(list, v)
+	}
+	return list, mapError(rows.Err())
+}
+func (r *BacktestRepository) Create(ctx context.Context, input model.BacktestInput, actor, requestID string) (model.BacktestRun, error) {
+	tx, e := r.pool.Begin(ctx)
+	if e != nil {
+		return model.BacktestRun{}, mapError(e)
+	}
+	defer tx.Rollback(context.Background())
+	var p model.StrategyParameters
+	var raw []byte
+	if e := tx.QueryRow(ctx, `SELECT parameters FROM veyra_strategy_version WHERE id=$1`, input.StrategyVersionID).Scan(&raw); e != nil {
+		return model.BacktestRun{}, mapError(e)
+	}
+	if json.Unmarshal(raw, &p) != nil {
+		return model.BacktestRun{}, apperror.ErrUnavailable
+	}
+	if len(input.Bars) < p.LongWindow+2 {
+		return model.BacktestRun{}, apperror.ErrInvalidInput
+	}
+	data, e := json.Marshal(model.ResearchInput{BacktestInput: input, Parameters: p})
+	if e != nil {
+		return model.BacktestRun{}, apperror.ErrInvalidInput
+	}
+	hash := sha256.Sum256(data)
+	id := uuid.NewString()
+	_, e = tx.Exec(ctx, `INSERT INTO veyra_backtest_run(id,strategy_version_id,input,input_hash,author_id) VALUES($1,$2,$3,$4,$5)`, id, input.StrategyVersionID, data, hex.EncodeToString(hash[:]), actor)
+	if e != nil {
+		return model.BacktestRun{}, mapError(e)
+	}
+	if e := appendAudit(ctx, tx, actor, "BACKTEST_QUEUED", "backtest_run", id, requestID, map[string]any{"strategy_version_id": input.StrategyVersionID, "input_hash": hex.EncodeToString(hash[:]), "bars": len(input.Bars)}); e != nil {
+		return model.BacktestRun{}, e
+	}
+	v, e := scanBacktest(tx.QueryRow(ctx, `SELECT `+backtestColumns+` FROM veyra_backtest_run WHERE id=$1`, id))
+	if e != nil {
+		return v, e
+	}
+	return v, mapError(tx.Commit(ctx))
+}
+func (r *BacktestRepository) Claim(ctx context.Context) (model.BacktestRun, error) {
+	return scanBacktest(r.pool.QueryRow(ctx, `WITH next AS (SELECT id FROM veyra_backtest_run WHERE status='PENDING' OR (status='RUNNING' AND lease_expires_at<now()) ORDER BY created_at,id LIMIT 1 FOR UPDATE SKIP LOCKED) UPDATE veyra_backtest_run r SET status='RUNNING',lease_id=$1,lease_expires_at=now()+interval '60 seconds' FROM next WHERE r.id=next.id RETURNING r.id,r.strategy_version_id,r.status,r.input,r.input_hash,r.result,r.failure,r.lease_id,r.created_at,r.finished_at`, uuid.NewString()))
+}
+func (r *BacktestRepository) Finish(ctx context.Context, job model.BacktestRun, result json.RawMessage, failure string) error {
+	status := "COMPLETED"
+	if failure != "" {
+		status = "FAILED"
+		result = nil
+	}
+	tag, e := r.pool.Exec(ctx, `UPDATE veyra_backtest_run SET status=$3,result=$4,failure=$5,lease_id=NULL,lease_expires_at=NULL,finished_at=now() WHERE id=$1 AND lease_id=$2 AND status='RUNNING' AND lease_expires_at>now()`, job.ID, job.LeaseID, status, result, failure)
+	if e != nil {
+		return mapError(e)
+	}
+	if tag.RowsAffected() != 1 {
+		return apperror.ErrConflict
+	}
+	return nil
+}

internal/repository/candidate.go

@@ -0,0 +1,205 @@
+package repository
+
+import (
+	"context"
+	"encoding/json"
+
+	"time"
+
+	"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"
+)
+
+type CandidateRepository struct{ pool *pgxpool.Pool }
+
+func NewCandidateRepository(pool *pgxpool.Pool) *CandidateRepository {
+	return &CandidateRepository{pool: pool}
+}
+
+const candidateColumns = `r.id,r.candidate_id,r.revision,r.version,r.status,r.data_mode,r.snapshot,r.author_id,r.observed_at,r.valid_until,r.published_at,r.created_at,r.updated_at,(r.revision=c.latest_revision),COALESCE(c.published_revision_id=r.id AND r.status='PUBLISHED' AND r.valid_until>now(),false),COALESCE((SELECT reason FROM veyra_candidate_review v WHERE v.revision_id=r.id AND v.decision='WITHDRAW' ORDER BY v.input_version DESC LIMIT 1),'')`
+
+func scanCandidate(row pgx.Row) (model.CandidateRevision, error) {
+	var r model.CandidateRevision
+	var snapshot []byte
+	err := row.Scan(&r.ID, &r.CandidateID, &r.Revision, &r.Version, &r.Status, &r.DataMode, &snapshot, &r.AuthorID, &r.ObservedAt, &r.ValidUntil, &r.PublishedAt, &r.CreatedAt, &r.UpdatedAt, &r.IsLatest, &r.IsActivePublication, &r.WithdrawalReason)
+	if err != nil {
+		return r, mapError(err)
+	}
+	if err := json.Unmarshal(snapshot, &r.Snapshot); err != nil {
+		return r, apperror.ErrUnavailable
+	}
+	return r, nil
+}
+
+const candidateJoin = ` FROM veyra_candidate_revision r JOIN veyra_candidate c ON c.id=r.candidate_id `
+
+func (r *CandidateRepository) List(ctx context.Context, admin bool, status string, limit int) ([]model.CandidateRevision, error) {
+	query := `SELECT ` + candidateColumns + candidateJoin + ` WHERE ($1 AND (r.revision=c.latest_revision OR c.published_revision_id=r.id) AND ($2='' OR r.status=$2)) OR (NOT $1 AND c.published_revision_id=r.id AND r.status='PUBLISHED' AND r.valid_until>now()) ORDER BY r.created_at DESC,r.id DESC LIMIT $3`
+	rows, err := r.pool.Query(ctx, query, admin, status, limit)
+	if err != nil {
+		return nil, mapError(err)
+	}
+	defer rows.Close()
+	list := make([]model.CandidateRevision, 0)
+	for rows.Next() {
+		v, err := scanCandidate(rows)
+		if err != nil {
+			return nil, err
+		}
+		list = append(list, v)
+	}
+	return list, mapError(rows.Err())
+}
+func (r *CandidateRepository) Detail(ctx context.Context, id string, admin bool) (model.CandidateDetail, error) {
+	revision, err := scanCandidate(r.pool.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE r.id=$1 AND ($2 OR r.published_at IS NOT NULL)`, id, admin))
+	detail := model.CandidateDetail{CandidateRevision: revision, Reviews: make([]model.CandidateReview, 0)}
+	if err != nil {
+		return detail, err
+	}
+	// App readers receive the publication, not internal reviewer identities or deliberations.
+	if !admin {
+		return detail, nil
+	}
+	rows, err := r.pool.Query(ctx, `SELECT id,revision_id,reviewer_id,decision,input_version,reason,created_at FROM veyra_candidate_review WHERE revision_id=$1 ORDER BY input_version`, id)
+	if err != nil {
+		return detail, mapError(err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var v model.CandidateReview
+		if err := rows.Scan(&v.ID, &v.RevisionID, &v.ReviewerID, &v.Decision, &v.InputVersion, &v.Reason, &v.CreatedAt); err != nil {
+			return detail, mapError(err)
+		}
+		detail.Reviews = append(detail.Reviews, v)
+	}
+	return detail, mapError(rows.Err())
+}
+
+// All mutations lock the candidate aggregate before its revisions, with the same lock order.
+func lockCandidate(ctx context.Context, tx pgx.Tx, source string) (model.CandidateRevision, error) {
+	var parent string
+	if err := tx.QueryRow(ctx, `SELECT candidate_id FROM veyra_candidate_revision WHERE id=$1`, source).Scan(&parent); err != nil {
+		return model.CandidateRevision{}, mapError(err)
+	}
+	var locked string
+	if err := tx.QueryRow(ctx, `SELECT id FROM veyra_candidate WHERE id=$1 FOR UPDATE`, parent).Scan(&locked); err != nil {
+		return model.CandidateRevision{}, mapError(err)
+	}
+	return scanCandidate(tx.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE r.id=$1 FOR UPDATE OF r`, source))
+}
+func (r *CandidateRepository) Create(ctx context.Context, input model.CandidateDraftInput, actor, requestID, source string, expected int64) (model.CandidateRevision, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return model.CandidateRevision{}, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	parent, revision := uuid.NewString(), 1
+	if source == "" {
+		_, err = tx.Exec(ctx, `INSERT INTO veyra_candidate(id,latest_revision) VALUES($1,1)`, parent)
+	} else {
+		previous, e := lockCandidate(ctx, tx, source)
+		if e != nil {
+			return model.CandidateRevision{}, e
+		}
+		if previous.Version != expected || !previous.IsLatest || previous.Status == "PENDING_REVIEW" {
+			return model.CandidateRevision{}, apperror.ErrConflict
+		}
+		parent, revision = previous.CandidateID, previous.Revision+1
+		_, err = tx.Exec(ctx, `UPDATE veyra_candidate SET latest_revision=$2 WHERE id=$1`, parent, revision)
+	}
+	if err != nil {
+		return model.CandidateRevision{}, mapError(err)
+	}
+	id := uuid.NewString()
+	snapshot, err := json.Marshal(input.Content)
+	if err != nil {
+		return model.CandidateRevision{}, apperror.ErrInvalidInput
+	}
+	_, err = tx.Exec(ctx, `INSERT INTO veyra_candidate_revision(id,candidate_id,revision,status,data_mode,snapshot,author_id,observed_at,valid_until) VALUES($1,$2,$3,'DRAFT','simulation',$4,$5,$6,$7)`, id, parent, revision, snapshot, actor, input.ObservedAt, input.ValidUntil)
+	if err != nil {
+		return model.CandidateRevision{}, mapError(err)
+	}
+	if err := appendAudit(ctx, tx, actor, "CANDIDATE_DRAFT_CREATED", "candidate_revision", id, requestID, map[string]any{"revision": revision, "source_revision_id": source}); err != nil {
+		return model.CandidateRevision{}, err
+	}
+	created, err := scanCandidate(tx.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE r.id=$1`, id))
+	if err != nil {
+		return created, err
+	}
+	return created, mapError(tx.Commit(ctx))
+}
+func appendCandidateEvent(ctx context.Context, tx pgx.Tx, r model.CandidateRevision, eventType string, details any) error {
+	payload, err := json.Marshal(details)
+	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,'candidate_revision',$2,$3,$4,$5)`, uuid.NewString(), r.ID, r.Version, eventType, payload)
+	return mapError(err)
+}
+func (r *CandidateRepository) Transition(ctx context.Context, id, action string, input model.CandidateActionInput, actor, requestID string) (model.CandidateRevision, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return model.CandidateRevision{}, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	revision, err := lockCandidate(ctx, tx, id)
+	if err != nil {
+		return revision, err
+	}
+	if revision.Version != input.ExpectedVersion {
+		return revision, apperror.ErrConflict
+	}
+	status, err := model.NextStatus(revision, action, time.Now().UTC())
+	if err != nil {
+		return revision, err
+	}
+	if action == "PUBLISH" {
+		// Superseded publications keep their historical content, with an invalidation event.
+		previous, err := scanCandidate(tx.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE c.id=$1 AND c.published_revision_id=r.id`, revision.CandidateID))
+		if err != nil && err != apperror.ErrNotFound {
+			return revision, err
+		}
+		if err == nil {
+			if err := appendCandidateEvent(ctx, tx, previous, "candidate.superseded", map[string]any{"replacement_revision_id": id}); err != nil {
+				return revision, err
+			}
+		}
+		_, err = tx.Exec(ctx, `UPDATE veyra_candidate SET published_revision_id=$2 WHERE id=$1`, revision.CandidateID, id)
+		if err != nil {
+			return revision, mapError(err)
+		}
+	} else if action == "WITHDRAW" {
+		_, err = tx.Exec(ctx, `UPDATE veyra_candidate SET published_revision_id=NULL WHERE id=$1`, revision.CandidateID)
+		if err != nil {
+			return revision, mapError(err)
+		}
+	}
+	_, err = tx.Exec(ctx, `UPDATE veyra_candidate_revision SET status=$2::varchar(24),version=version+1,updated_at=now(),published_at=CASE WHEN $2::varchar(24)='PUBLISHED' THEN now() ELSE published_at END WHERE id=$1`, id, status)
+	if err != nil {
+		return revision, mapError(err)
+	}
+	_, err = tx.Exec(ctx, `INSERT INTO veyra_candidate_review(id,revision_id,reviewer_id,decision,input_version,reason) VALUES($1,$2,$3,$4,$5,$6)`, uuid.NewString(), id, actor, action, input.ExpectedVersion, input.Reason)
+	if err != nil {
+		return revision, mapError(err)
+	}
+	if err := appendAudit(ctx, tx, actor, "CANDIDATE_"+action, "candidate_revision", id, requestID, map[string]any{"reason": input.Reason, "input_version": input.ExpectedVersion, "status": status}); err != nil {
+		return revision, err
+	}
+	revision, err = scanCandidate(tx.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE r.id=$1`, id))
+	if err != nil {
+		return revision, err
+	}
+	if action == "PUBLISH" || action == "WITHDRAW" {
+		eventType := "candidate.published"
+		if action == "WITHDRAW" {
+			eventType = "candidate.withdrawn"
+		}
+		if err := appendCandidateEvent(ctx, tx, revision, eventType, map[string]any{"revision_id": id, "version": revision.Version, "reason": input.Reason}); err != nil {
+			return revision, err
+		}
+	}
+	return revision, mapError(tx.Commit(ctx))
+}

internal/repository/identity.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 
@@ -45,9 +46,25 @@ func (r *IdentityRepository) FindCredential(ctx context.Context, login string) (
 	return u, c, mapError(err)
 }
 
-func (r *IdentityRepository) CreateSession(ctx context.Context, s model.Session) error {
-	_, err := r.pool.Exec(ctx, `INSERT INTO veyra_session(id,user_id,token_hash,audience,expires_at) VALUES($1,$2,$3,$4,$5)`, s.ID, s.UserID, s.TokenHash, s.Audience, s.ExpiresAt)
-	return mapError(err)
+func (r *IdentityRepository) CreateSession(ctx context.Context, s model.Session, verifiedPasswordHash string) error {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	var userID string
+	err = tx.QueryRow(ctx, `SELECT u.id FROM veyra_user u JOIN veyra_credential c ON c.user_id=u.id WHERE u.id=$1 AND c.password_hash=$2 AND u.status='ACTIVE' AND (($3='admin' AND u.role='ADMIN') OR ($3='app' AND u.role='USER')) FOR SHARE OF u,c`, s.UserID, verifiedPasswordHash, s.Audience).Scan(&userID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return apperror.ErrUnauthenticated
+	}
+	if err != nil {
+		return mapError(err)
+	}
+	_, err = tx.Exec(ctx, `INSERT INTO veyra_session(id,user_id,token_hash,audience,expires_at) VALUES($1,$2,$3,$4,$5)`, s.ID, s.UserID, s.TokenHash, s.Audience, s.ExpiresAt)
+	if err != nil {
+		return mapError(err)
+	}
+	return mapError(tx.Commit(ctx))
 }
 
 func (r *IdentityRepository) FindSessionUser(ctx context.Context, hash, audience string, now time.Time) (model.User, error) {
@@ -80,3 +97,33 @@ func (r *IdentityRepository) ListUsers(ctx context.Context, limit int) ([]model.
 	}
 	return users, mapError(rows.Err())
 }
+
+func (r *IdentityRepository) ChangeAdminCredential(ctx context.Context, previousLogin, login, passwordHash string, now time.Time) (string, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return "", mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	var userID, role, status string
+	err = tx.QueryRow(ctx, `SELECT u.id,u.role,u.status FROM veyra_user u JOIN veyra_credential c ON c.user_id=u.id WHERE c.login_name=$1 FOR UPDATE OF u,c`, previousLogin).Scan(&userID, &role, &status)
+	if err != nil {
+		return "", mapError(err)
+	}
+	if role != "ADMIN" || status != "ACTIVE" {
+		return "", apperror.ErrForbidden
+	}
+	if _, err := tx.Exec(ctx, `UPDATE veyra_credential SET login_name=$2,password_hash=$3 WHERE user_id=$1`, userID, login, passwordHash); err != nil {
+		return "", mapError(err)
+	}
+	tag, err := tx.Exec(ctx, `UPDATE veyra_session SET revoked_at=$2 WHERE user_id=$1 AND revoked_at IS NULL`, userID, now)
+	if err != nil {
+		return "", mapError(err)
+	}
+	if err := appendAudit(ctx, tx, userID, "ADMIN_CREDENTIAL_CHANGED", "user", userID, uuid.NewString(), map[string]any{"sessions_revoked": tag.RowsAffected()}); err != nil {
+		return "", err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return "", mapError(err)
+	}
+	return userID, nil
+}

internal/repository/market.go

@@ -0,0 +1,78 @@
+package repository
+
+import (
+	"context"
+	"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
+	"git.sundynix.cn/Blizzard/Veyra/internal/model"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"time"
+)
+
+type MarketRepository struct{ pool *pgxpool.Pool }
+
+func NewMarketRepository(pool *pgxpool.Pool) *MarketRepository { return &MarketRepository{pool} }
+func (r *MarketRepository) List(ctx context.Context, query string) ([]model.Instrument, error) {
+	rows, e := r.pool.Query(ctx, `SELECT id,symbol,exchange,name,price_cents,observed_at,received_at,source_id,source_version FROM veyra_instrument WHERE symbol LIKE $1 OR name LIKE $1 ORDER BY (price_cents>0) DESC,symbol LIMIT 100`, "%"+query+"%")
+	if e != nil {
+		return nil, mapError(e)
+	}
+	defer rows.Close()
+	list := make([]model.Instrument, 0)
+	for rows.Next() {
+		var v model.Instrument
+		if e := rows.Scan(&v.ID, &v.Symbol, &v.Exchange, &v.Name, &v.PriceCents, &v.ObservedAt, &v.ReceivedAt, &v.SourceID, &v.SourceVersion); e != nil {
+			return nil, mapError(e)
+		}
+		v.Price = model.FormatMoney(v.PriceCents)
+		list = append(list, v)
+	}
+	return list, mapError(rows.Err())
+}
+func (r *MarketRepository) Save(ctx context.Context, c model.ProviderConfig, list []model.Instrument, started time.Time, message string) error {
+	tx, e := r.pool.Begin(ctx)
+	if e != nil {
+		return mapError(e)
+	}
+	defer tx.Rollback(context.Background())
+	var version int64
+	var enabled bool
+	if e := tx.QueryRow(ctx, `SELECT version,enabled FROM veyra_provider_config WHERE id=$1 FOR SHARE`, c.ID).Scan(&version, &enabled); e != nil {
+		return mapError(e)
+	}
+	if !enabled || version != c.Version {
+		return apperror.ErrConflict
+	}
+	status := "COMPLETED"
+	if message != "" {
+		status = "FAILED"
+		list = nil
+	}
+	for _, v := range list {
+		_, e := tx.Exec(ctx, `INSERT INTO veyra_instrument(id,symbol,exchange,name,price_cents,observed_at,source_id,source_version) VALUES($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT(exchange,symbol) DO UPDATE SET name=excluded.name,price_cents=excluded.price_cents,observed_at=excluded.observed_at,received_at=now(),source_id=excluded.source_id,source_version=excluded.source_version WHERE veyra_instrument.observed_at<=excluded.observed_at`, uuid.NewString(), v.Symbol, v.Exchange, v.Name, v.PriceCents, v.ObservedAt, c.ID, c.Version)
+		if e != nil {
+			return mapError(e)
+		}
+	}
+	_, e = tx.Exec(ctx, `INSERT INTO veyra_market_sync(id,source_id,source_version,instrument_count,status,message,started_at) VALUES($1,$2,$3,$4,$5,$6,$7)`, uuid.NewString(), c.ID, c.Version, len(list), status, message, started)
+	if e != nil {
+		return mapError(e)
+	}
+	return mapError(tx.Commit(ctx))
+}
+func (r *MarketRepository) Tasks(ctx context.Context) ([]model.MarketSync, error) {
+	rows, e := r.pool.Query(ctx, `SELECT id,source_id,source_version,instrument_count,status,message,started_at,finished_at FROM veyra_market_sync ORDER BY finished_at DESC,id DESC LIMIT 20`)
+	if e != nil {
+		return nil, mapError(e)
+	}
+	defer rows.Close()
+	list := make([]model.MarketSync, 0)
+	for rows.Next() {
+		var v model.MarketSync
+		if e := rows.Scan(&v.ID, &v.SourceID, &v.SourceVersion, &v.InstrumentCount, &v.Status, &v.Message, &v.StartedAt, &v.FinishedAt); e != nil {
+			return nil, mapError(e)
+		}
+		list = append(list, v)
+	}
+	return list, mapError(rows.Err())
+}

internal/repository/notification.go

@@ -0,0 +1,281 @@
+package repository
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"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"
+	"time"
+)
+
+type NotificationRepository struct{ pool *pgxpool.Pool }
+
+func NewNotificationRepository(pool *pgxpool.Pool) *NotificationRepository {
+	return &NotificationRepository{pool}
+}
+func (r *NotificationRepository) List(ctx context.Context, user string) ([]model.Notification, error) {
+	rows, err := r.pool.Query(ctx, `SELECT id,user_id,event_id,plan_id,title,body,read_at,created_at FROM veyra_notification WHERE user_id=$1 ORDER BY created_at DESC,id DESC LIMIT 100`, user)
+	if err != nil {
+		return nil, mapError(err)
+	}
+	defer rows.Close()
+	list := make([]model.Notification, 0)
+	for rows.Next() {
+		var n model.Notification
+		if err := rows.Scan(&n.ID, &n.UserID, &n.EventID, &n.PlanID, &n.Title, &n.Body, &n.ReadAt, &n.CreatedAt); err != nil {
+			return nil, mapError(err)
+		}
+		list = append(list, n)
+	}
+	return list, mapError(rows.Err())
+}
+func (r *NotificationRepository) MarkRead(ctx context.Context, user, id string) error {
+	tag, err := r.pool.Exec(ctx, `UPDATE veyra_notification SET read_at=COALESCE(read_at,now()) WHERE id=$1 AND user_id=$2`, id, user)
+	if err != nil {
+		return mapError(err)
+	}
+	if tag.RowsAffected() != 1 {
+		return apperror.ErrNotFound
+	}
+	return nil
+}
+func (r *NotificationRepository) Tasks(ctx context.Context) ([]model.NotificationTask, error) {
+	rows, err := r.pool.Query(ctx, `SELECT e.id,e.aggregate_type,e.aggregate_id,e.aggregate_version,e.event_type,e.payload,e.created_at,e.processed_at,(SELECT count(*)::integer FROM veyra_notification n WHERE n.event_id=e.id) FROM veyra_outbox_event e ORDER BY created_at DESC,id DESC LIMIT 100`)
+	if err != nil {
+		return nil, mapError(err)
+	}
+	defer rows.Close()
+	list := make([]model.NotificationTask, 0)
+	for rows.Next() {
+		var t model.NotificationTask
+		if err := rows.Scan(&t.ID, &t.AggregateType, &t.AggregateID, &t.AggregateVersion, &t.EventType, &t.Payload, &t.CreatedAt, &t.ProcessedAt, &t.RecipientCount); err != nil {
+			return nil, mapError(err)
+		}
+		list = append(list, t)
+	}
+	return list, mapError(rows.Err())
+}
+
+const quoteColumns = `id,symbol,exchange,data_mode,price_minor,observed_at,version,created_at,updated_at`
+
+func scanQuote(row pgx.Row) (model.QuoteSnapshot, error) {
+	var q model.QuoteSnapshot
+	err := row.Scan(&q.ID, &q.Symbol, &q.Exchange, &q.DataMode, &q.PriceMinor, &q.ObservedAt, &q.Version, &q.CreatedAt, &q.UpdatedAt)
+	q.Price = model.FormatMoney(q.PriceMinor)
+	return q, mapError(err)
+}
+func (r *NotificationRepository) Quotes(ctx context.Context) ([]model.QuoteSnapshot, error) {
+	rows, err := r.pool.Query(ctx, `SELECT `+quoteColumns+` FROM veyra_quote_snapshot ORDER BY updated_at DESC,id DESC LIMIT 100`)
+	if err != nil {
+		return nil, mapError(err)
+	}
+	defer rows.Close()
+	list := make([]model.QuoteSnapshot, 0)
+	for rows.Next() {
+		q, err := scanQuote(rows)
+		if err != nil {
+			return nil, err
+		}
+		list = append(list, q)
+	}
+	return list, mapError(rows.Err())
+}
+func (r *NotificationRepository) SaveQuote(ctx context.Context, q model.QuoteSnapshot, actor, requestID string) (model.QuoteSnapshot, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return q, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	q, err = scanQuote(tx.QueryRow(ctx, `INSERT INTO veyra_quote_snapshot(id,symbol,exchange,data_mode,price_minor,observed_at,version) VALUES($1,$2,$3,'simulation',$4,$5,1) ON CONFLICT(exchange,symbol,data_mode) DO UPDATE SET price_minor=EXCLUDED.price_minor,observed_at=EXCLUDED.observed_at,version=veyra_quote_snapshot.version+1,updated_at=now() WHERE EXCLUDED.observed_at>veyra_quote_snapshot.observed_at RETURNING `+quoteColumns, uuid.NewString(), q.Symbol, q.Exchange, q.PriceMinor, q.ObservedAt))
+	if err == apperror.ErrNotFound {
+		return q, apperror.ErrConflict
+	}
+	if err != nil {
+		return q, err
+	}
+	payload, err := json.Marshal(q)
+	if err != nil {
+		return q, apperror.ErrUnavailable
+	}
+	_, err = tx.Exec(ctx, `INSERT INTO veyra_outbox_event(id,aggregate_type,aggregate_id,aggregate_version,event_type,payload) VALUES($1,'quote_snapshot',$2,$3,'quote.observed',$4)`, uuid.NewString(), q.ID, q.Version, payload)
+	if err != nil {
+		return q, mapError(err)
+	}
+	if err := appendAudit(ctx, tx, actor, "SIMULATION_QUOTE_RECORDED", "quote_snapshot", q.ID, requestID, map[string]any{"symbol": q.Symbol, "exchange": q.Exchange, "version": q.Version}); err != nil {
+		return q, err
+	}
+	return q, mapError(tx.Commit(ctx))
+}
+func insertNotification(ctx context.Context, tx pgx.Tx, user, event string, plan *string, title, body string) error {
+	_, err := tx.Exec(ctx, `INSERT INTO veyra_notification(id,user_id,event_id,plan_id,title,body) VALUES($1,$2,$3,$4,$5,$6) ON CONFLICT DO NOTHING`, uuid.NewString(), user, event, plan, title, body)
+	return mapError(err)
+}
+func processQuote(ctx context.Context, tx pgx.Tx, event model.OutboxEvent) error {
+	var payload model.QuoteSnapshot
+	if json.Unmarshal(event.Payload, &payload) != nil {
+		return apperror.ErrUnavailable
+	}
+	// A stale or superseded snapshot cannot generate a delayed buy reminder.
+	current, err := scanQuote(tx.QueryRow(ctx, `SELECT `+quoteColumns+` FROM veyra_quote_snapshot WHERE id=$1 FOR SHARE`, event.AggregateID))
+	if err != nil {
+		return err
+	}
+	if current.Version != event.AggregateVersion || time.Since(current.ObservedAt) > time.Minute || current.ObservedAt.After(time.Now().Add(time.Minute)) {
+		return nil
+	}
+	rows, err := tx.Query(ctx, `SELECT p.id,p.candidate_revision_id,a.user_id FROM veyra_trade_plan p JOIN veyra_candidate_revision r ON r.id=p.candidate_revision_id JOIN veyra_account a ON a.id=p.account_id JOIN veyra_user u ON u.id=a.user_id WHERE p.status IN ('ACTIVE','PART_FILLED') AND u.status='ACTIVE' AND r.snapshot->>'symbol'=$1 AND r.snapshot->>'exchange'=$2 ORDER BY r.candidate_id,p.id`, current.Symbol, current.Exchange)
+	if err != nil {
+		return mapError(err)
+	}
+	type target struct{ plan, revision, user string }
+	targets := make([]target, 0)
+	for rows.Next() {
+		var t target
+		if err := rows.Scan(&t.plan, &t.revision, &t.user); err != nil {
+			rows.Close()
+			return mapError(err)
+		}
+		targets = append(targets, t)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return mapError(err)
+	}
+	for _, target := range targets {
+		publication, err := lockCandidate(ctx, tx, target.revision)
+		if err != nil {
+			return err
+		}
+		if !publication.IsActivePublication {
+			continue
+		}
+		var status string
+		var valid time.Time
+		if err := tx.QueryRow(ctx, `SELECT status,valid_until FROM veyra_trade_plan WHERE id=$1 FOR UPDATE`, target.plan).Scan(&status, &valid); err != nil {
+			return mapError(err)
+		}
+		if (status != "ACTIVE" && status != "PART_FILLED") || !valid.After(time.Now()) {
+			continue
+		}
+		low, e1 := model.ParseMoney(publication.Snapshot.ReferencePlan.EntryLow)
+		high, e2 := model.ParseMoney(publication.Snapshot.ReferencePlan.EntryHigh)
+		if e1 != nil || e2 != nil {
+			return apperror.ErrUnavailable
+		}
+		if current.PriceMinor < low || current.PriceMinor > high {
+			continue
+		}
+		tag, err := tx.Exec(ctx, `INSERT INTO veyra_plan_trigger(id,plan_id,quote_snapshot_id,quote_version) VALUES($1,$2,$3,$4) ON CONFLICT(plan_id) DO NOTHING`, uuid.NewString(), target.plan, current.ID, current.Version)
+		if err != nil {
+			return mapError(err)
+		}
+		if tag.RowsAffected() == 0 {
+			continue
+		}
+		body := fmt.Sprintf("模拟价格 ¥%s 进入 %s 的参考区间。观察时间 %s。请打开计划核对当前状态；此提醒不代表成交。", current.Price, publication.Snapshot.Name, current.ObservedAt.Format(time.RFC3339))
+		if err := insertNotification(ctx, tx, target.user, event.ID, &target.plan, "模拟计划条件已满足", body); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+func processCandidateInvalidation(ctx context.Context, tx pgx.Tx, event model.OutboxEvent) error {
+	rows, err := tx.Query(ctx, `SELECT p.id,a.user_id FROM veyra_trade_plan p JOIN veyra_account a ON a.id=p.account_id WHERE p.candidate_revision_id=$1 AND p.status IN ('ACTIVE','PART_FILLED') ORDER BY p.id`, event.AggregateID)
+	if err != nil {
+		return mapError(err)
+	}
+	type target struct{ plan, user string }
+	targets := make([]target, 0)
+	for rows.Next() {
+		var t target
+		if err := rows.Scan(&t.plan, &t.user); err != nil {
+			rows.Close()
+			return mapError(err)
+		}
+		targets = append(targets, t)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return mapError(err)
+	}
+	for _, t := range targets {
+		tag, err := tx.Exec(ctx, `UPDATE veyra_trade_plan SET status='INVALIDATED',reserved_minor=0,version=version+1,updated_at=now() WHERE id=$1 AND status IN ('ACTIVE','PART_FILLED')`, t.plan)
+		if err != nil {
+			return mapError(err)
+		}
+		reason := "对应候选已撤回或被新版替代，保留预算已释放。已记录成交与持仓保留。"
+		if err := insertNotification(ctx, tx, t.user, event.ID, &t.plan, "计划已失效", reason); err != nil {
+			return err
+		}
+		if tag.RowsAffected() > 0 {
+			if err := appendAudit(ctx, tx, t.user, "PLAN_INVALIDATED", "trade_plan", t.plan, event.ID, map[string]any{"cause": event.EventType}); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+func (r *NotificationRepository) ProcessNext(ctx context.Context) (bool, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return false, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	var event model.OutboxEvent
+	err = tx.QueryRow(ctx, `SELECT id,aggregate_type,aggregate_id,aggregate_version,event_type,payload,created_at,processed_at FROM veyra_outbox_event WHERE processed_at IS NULL ORDER BY created_at,id LIMIT 1 FOR UPDATE SKIP LOCKED`).Scan(&event.ID, &event.AggregateType, &event.AggregateID, &event.AggregateVersion, &event.EventType, &event.Payload, &event.CreatedAt, &event.ProcessedAt)
+	if err == pgx.ErrNoRows {
+		rows, err := tx.Query(ctx, `SELECT DISTINCT a.user_id FROM veyra_trade_plan p JOIN veyra_account a ON a.id=p.account_id JOIN veyra_candidate_revision r ON r.id=p.candidate_revision_id JOIN veyra_candidate c ON c.id=r.candidate_id WHERE p.status IN ('ACTIVE','PART_FILLED') AND (p.valid_until<=now() OR r.valid_until<=now() OR r.status<>'PUBLISHED' OR c.published_revision_id IS DISTINCT FROM r.id) LIMIT 100`)
+		if err != nil {
+			return false, mapError(err)
+		}
+		users := make([]string, 0)
+		for rows.Next() {
+			var user string
+			if err := rows.Scan(&user); err != nil {
+				rows.Close()
+				return false, mapError(err)
+			}
+			users = append(users, user)
+		}
+		rows.Close()
+		if err := rows.Err(); err != nil {
+			return false, mapError(err)
+		}
+		for _, user := range users {
+			if err := invalidatePlans(ctx, tx, user); err != nil {
+				return false, err
+			}
+		}
+		return len(users) > 0, mapError(tx.Commit(ctx))
+	}
+	if err != nil {
+		return false, mapError(err)
+	}
+	switch event.EventType {
+	case "quote.observed":
+		err = processQuote(ctx, tx, event)
+	case "candidate.withdrawn", "candidate.superseded":
+		err = processCandidateInvalidation(ctx, tx, event)
+	case "plan.invalidated":
+		var user string
+		var plan string
+		err = tx.QueryRow(ctx, `SELECT a.user_id,p.id FROM veyra_trade_plan p JOIN veyra_account a ON a.id=p.account_id WHERE p.id=$1`, event.AggregateID).Scan(&user, &plan)
+		if err == nil {
+			err = insertNotification(ctx, tx, user, event.ID, &plan, "计划已失效", "对应候选已失效或计划已过期，保留预算已释放。已记录成交与持仓保留。")
+		}
+	case "candidate.published", "position.changed":
+	default:
+		return false, apperror.ErrInvalidInput
+	}
+	if err != nil {
+		return false, err
+	}
+	if _, err := tx.Exec(ctx, `UPDATE veyra_outbox_event SET processed_at=now() WHERE id=$1`, event.ID); err != nil {
+		return false, mapError(err)
+	}
+	return true, mapError(tx.Commit(ctx))
+}

internal/repository/provider.go

@@ -0,0 +1,140 @@
+package repository
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
+	"git.sundynix.cn/Blizzard/Veyra/internal/model"
+)
+
+type ProviderRepository struct{ pool *pgxpool.Pool }
+
+func NewProviderRepository(pool *pgxpool.Pool) *ProviderRepository {
+	return &ProviderRepository{pool: pool}
+}
+
+const providerColumns = `id,config_key,name,base_url,model_name,enabled,timeout_seconds,encrypted_api_key,version,last_test_status,last_tested_at,last_test_message,updated_by,created_at,updated_at,wire_api,model_catalog_url,disable_response_storage,requires_openai_auth,market_adapter,history_base_url,quote_base_url`
+
+func scanProvider(row pgx.Row) (model.ProviderConfig, error) {
+	var c model.ProviderConfig
+	err := row.Scan(&c.ID, &c.ConfigKey, &c.Name, &c.BaseURL, &c.ModelName, &c.Enabled, &c.TimeoutSeconds, &c.EncryptedAPIKey, &c.Version, &c.LastTestStatus, &c.LastTestedAt, &c.LastTestMessage, &c.UpdatedBy, &c.CreatedAt, &c.UpdatedAt, &c.WireAPI, &c.ModelCatalogURL, &c.DisableResponseStorage, &c.RequiresOpenAIAuth, &c.MarketAdapter, &c.HistoryBaseURL, &c.QuoteBaseURL)
+	c.HasAPIKey = len(c.EncryptedAPIKey) > 0
+	return c, mapError(err)
+}
+func (r *ProviderRepository) Get(ctx context.Context, key string) (model.ProviderConfig, error) {
+	return scanProvider(r.pool.QueryRow(ctx, `SELECT `+providerColumns+` FROM veyra_provider_config WHERE config_key=$1`, key))
+}
+func (r *ProviderRepository) List(ctx context.Context) ([]model.ProviderConfig, error) {
+	rows, err := r.pool.Query(ctx, `SELECT `+providerColumns+` FROM veyra_provider_config ORDER BY config_key`)
+	if err != nil {
+		return nil, mapError(err)
+	}
+	defer rows.Close()
+	configs := make([]model.ProviderConfig, 0)
+	for rows.Next() {
+		c, err := scanProvider(rows)
+		if err != nil {
+			return nil, err
+		}
+		configs = append(configs, c)
+	}
+	return configs, mapError(rows.Err())
+}
+
+func (r *ProviderRepository) Save(ctx context.Context, c model.ProviderConfig, expected int64, actor, requestID string, secretChanged bool) (model.ProviderConfig, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return model.ProviderConfig{}, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	// A per-key lock also serializes two simultaneous first-time creates.
+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext($1)::bigint)`, "veyra:provider:"+c.ConfigKey); err != nil {
+		return model.ProviderConfig{}, mapError(err)
+	}
+	var version int64
+	err = tx.QueryRow(ctx, `SELECT version FROM veyra_provider_config WHERE config_key=$1`, c.ConfigKey).Scan(&version)
+	if err != nil && err != pgx.ErrNoRows {
+		return model.ProviderConfig{}, mapError(err)
+	}
+	if version != expected {
+		return model.ProviderConfig{}, apperror.ErrConflict
+	}
+	c.Version = version + 1
+	if version == 0 {
+		c.ID = uuid.NewString()
+		_, err = tx.Exec(ctx, `INSERT INTO veyra_provider_config(id,config_key,name,base_url,model_name,enabled,timeout_seconds,encrypted_api_key,version,last_test_status,updated_by,wire_api,model_catalog_url,disable_response_storage,requires_openai_auth,market_adapter,history_base_url,quote_base_url) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18)`, c.ID, c.ConfigKey, c.Name, c.BaseURL, c.ModelName, c.Enabled, c.TimeoutSeconds, c.EncryptedAPIKey, c.Version, c.LastTestStatus, actor, c.WireAPI, c.ModelCatalogURL, c.DisableResponseStorage, c.RequiresOpenAIAuth, c.MarketAdapter, c.HistoryBaseURL, c.QuoteBaseURL)
+	} else {
+		_, err = tx.Exec(ctx, `UPDATE veyra_provider_config SET name=$2,base_url=$3,model_name=$4,enabled=$5,timeout_seconds=$6,encrypted_api_key=$7,version=$8,last_test_status=$9,last_tested_at=NULL,last_test_message='',updated_by=$10,wire_api=$11,model_catalog_url=$12,disable_response_storage=$13,requires_openai_auth=$14,market_adapter=$15,history_base_url=$16,quote_base_url=$17,updated_at=now() WHERE config_key=$1`, c.ConfigKey, c.Name, c.BaseURL, c.ModelName, c.Enabled, c.TimeoutSeconds, c.EncryptedAPIKey, c.Version, c.LastTestStatus, actor, c.WireAPI, c.ModelCatalogURL, c.DisableResponseStorage, c.RequiresOpenAIAuth, c.MarketAdapter, c.HistoryBaseURL, c.QuoteBaseURL)
+	}
+	if err != nil {
+		return model.ProviderConfig{}, mapError(err)
+	}
+	c, err = scanProvider(tx.QueryRow(ctx, `SELECT `+providerColumns+` FROM veyra_provider_config WHERE config_key=$1`, c.ConfigKey))
+	if err != nil {
+		return c, err
+	}
+	snapshot, err := json.Marshal(c)
+	if err != nil {
+		return c, apperror.ErrUnavailable
+	}
+	if _, err := tx.Exec(ctx, `INSERT INTO veyra_provider_config_history(id,config_id,version,snapshot,encrypted_api_key,actor_id) VALUES($1,$2,$3,$4,$5,$6)`, uuid.NewString(), c.ID, c.Version, snapshot, c.EncryptedAPIKey, actor); err != nil {
+		return c, mapError(err)
+	}
+	if err := appendAudit(ctx, tx, actor, "PROVIDER_CONFIG_SAVED", "provider_config", c.ID, requestID, map[string]any{"config_key": c.ConfigKey, "version": c.Version, "enabled": c.Enabled, "secret_changed": secretChanged}); err != nil {
+		return c, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return c, mapError(err)
+	}
+	return c, nil
+}
+
+func (r *ProviderRepository) RecordTest(ctx context.Context, c model.ProviderConfig, status, message, actor, requestID string) error {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	tag, err := tx.Exec(ctx, `UPDATE veyra_provider_config SET last_test_status=$3,last_test_message=$4,last_tested_at=$5 WHERE id=$1 AND version=$2`, c.ID, c.Version, status, message, time.Now().UTC())
+	if err != nil {
+		return mapError(err)
+	}
+	if tag.RowsAffected() != 1 {
+		return apperror.ErrConflict
+	}
+	if err := appendAudit(ctx, tx, actor, "PROVIDER_TESTED", "provider_config", c.ID, requestID, map[string]any{"version": c.Version, "status": status}); err != nil {
+		return err
+	}
+	return mapError(tx.Commit(ctx))
+}
+
+func appendAudit(ctx context.Context, tx pgx.Tx, actor, action, objectType, objectID, requestID string, details any) error {
+	encoded, err := json.Marshal(details)
+	if err != nil {
+		return apperror.ErrUnavailable
+	}
+	_, err = tx.Exec(ctx, `INSERT INTO veyra_audit_event(id,actor_id,action,object_type,object_id,request_id,details) VALUES($1,$2,$3,$4,$5,$6,$7)`, uuid.NewString(), actor, action, objectType, objectID, requestID, encoded)
+	return mapError(err)
+}
+func (r *ProviderRepository) Audit(ctx context.Context) ([]model.AuditEvent, error) {
+	rows, err := r.pool.Query(ctx, `SELECT id,actor_id,action,object_type,object_id,request_id,details,created_at FROM veyra_audit_event ORDER BY created_at DESC,id DESC LIMIT 100`)
+	if err != nil {
+		return nil, mapError(err)
+	}
+	defer rows.Close()
+	list := make([]model.AuditEvent, 0)
+	for rows.Next() {
+		var a model.AuditEvent
+		if err := rows.Scan(&a.ID, &a.ActorID, &a.Action, &a.ObjectType, &a.ObjectID, &a.RequestID, &a.Details, &a.CreatedAt); err != nil {
+			return nil, mapError(err)
+		}
+		list = append(list, a)
+	}
+	return list, mapError(rows.Err())
+}

internal/repository/strategy.go

@@ -0,0 +1,153 @@
+package repository
+
+import (
+	"context"
+	"encoding/json"
+	"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
+	"git.sundynix.cn/Blizzard/Veyra/internal/model"
+	"git.sundynix.cn/Blizzard/Veyra/internal/service"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type StrategyRepository struct{ pool *pgxpool.Pool }
+
+func NewStrategyRepository(pool *pgxpool.Pool) *StrategyRepository { return &StrategyRepository{pool} }
+
+const strategyColumns = `id,code,name,status,latest_revision,active_version_id,version,created_at,updated_at`
+
+func scanStrategy(row pgx.Row) (model.Strategy, error) {
+	var s model.Strategy
+	err := row.Scan(&s.ID, &s.Code, &s.Name, &s.Status, &s.LatestRevision, &s.ActiveVersionID, &s.Version, &s.CreatedAt, &s.UpdatedAt)
+	return s, mapError(err)
+}
+func (r *StrategyRepository) List(ctx context.Context) ([]model.Strategy, error) {
+	rows, err := r.pool.Query(ctx, `SELECT `+strategyColumns+` FROM veyra_strategy ORDER BY updated_at DESC,id DESC LIMIT 100`)
+	if err != nil {
+		return nil, mapError(err)
+	}
+	defer rows.Close()
+	list := make([]model.Strategy, 0)
+	for rows.Next() {
+		s, err := scanStrategy(rows)
+		if err != nil {
+			return nil, err
+		}
+		list = append(list, s)
+	}
+	return list, mapError(rows.Err())
+}
+func (r *StrategyRepository) Detail(ctx context.Context, id string) (model.StrategyDetail, error) {
+	tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly})
+	if err != nil {
+		return model.StrategyDetail{}, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	return strategyDetail(ctx, tx, id)
+}
+func strategyDetail(ctx context.Context, tx pgx.Tx, id string) (model.StrategyDetail, error) {
+	s, err := scanStrategy(tx.QueryRow(ctx, `SELECT `+strategyColumns+` FROM veyra_strategy WHERE id=$1`, id))
+	detail := model.StrategyDetail{Strategy: s, Versions: make([]model.StrategyVersion, 0)}
+	if err != nil {
+		return detail, err
+	}
+	rows, err := tx.Query(ctx, `SELECT id,strategy_id,revision,algorithm,parameters,description,author_id,created_at FROM veyra_strategy_version WHERE strategy_id=$1 ORDER BY revision DESC`, id)
+	if err != nil {
+		return detail, mapError(err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var v model.StrategyVersion
+		var parameters []byte
+		if err := rows.Scan(&v.ID, &v.StrategyID, &v.Revision, &v.Algorithm, &parameters, &v.Description, &v.AuthorID, &v.CreatedAt); err != nil {
+			return detail, mapError(err)
+		}
+		if json.Unmarshal(parameters, &v.Parameters) != nil {
+			return detail, apperror.ErrUnavailable
+		}
+		detail.Versions = append(detail.Versions, v)
+	}
+	return detail, mapError(rows.Err())
+}
+func (r *StrategyRepository) Save(ctx context.Context, id string, input service.StrategyInput, actor, requestID string) (model.StrategyDetail, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return model.StrategyDetail{}, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	revision := 1
+	if id == "" {
+		id = uuid.NewString()
+		_, err = tx.Exec(ctx, `INSERT INTO veyra_strategy(id,code,name) VALUES($1,$2,$3)`, id, input.Code, input.Name)
+	} else {
+		s, e := scanStrategy(tx.QueryRow(ctx, `SELECT `+strategyColumns+` FROM veyra_strategy WHERE id=$1 FOR UPDATE`, id))
+		if e != nil {
+			return model.StrategyDetail{}, e
+		}
+		if s.Version != input.ExpectedVersion || s.Code != input.Code {
+			return model.StrategyDetail{}, apperror.ErrConflict
+		}
+		revision = s.LatestRevision + 1
+		_, err = tx.Exec(ctx, `UPDATE veyra_strategy SET name=$2,latest_revision=$3,version=version+1,updated_at=now() WHERE id=$1`, id, input.Name, revision)
+	}
+	if err != nil {
+		return model.StrategyDetail{}, mapError(err)
+	}
+	parameters, err := json.Marshal(input.Parameters)
+	if err != nil {
+		return model.StrategyDetail{}, apperror.ErrInvalidInput
+	}
+	versionID := uuid.NewString()
+	_, err = tx.Exec(ctx, `INSERT INTO veyra_strategy_version(id,strategy_id,revision,algorithm,parameters,description,author_id) VALUES($1,$2,$3,'sma_cross',$4,$5,$6)`, versionID, id, revision, parameters, input.Description, actor)
+	if err != nil {
+		return model.StrategyDetail{}, mapError(err)
+	}
+	if err := appendAudit(ctx, tx, actor, "STRATEGY_VERSION_CREATED", "strategy", id, requestID, map[string]any{"revision": revision, "version_id": versionID}); err != nil {
+		return model.StrategyDetail{}, err
+	}
+	detail, err := strategyDetail(ctx, tx, id)
+	if err != nil {
+		return detail, err
+	}
+	return detail, mapError(tx.Commit(ctx))
+}
+func (r *StrategyRepository) Deploy(ctx context.Context, id, versionID, status string, expected int64, reason, actor, requestID string) (model.StrategyDetail, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return model.StrategyDetail{}, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	s, err := scanStrategy(tx.QueryRow(ctx, `SELECT `+strategyColumns+` FROM veyra_strategy WHERE id=$1 FOR UPDATE`, id))
+	if err != nil {
+		return model.StrategyDetail{}, err
+	}
+	if s.Version != expected {
+		return model.StrategyDetail{}, apperror.ErrConflict
+	}
+	var selected *string
+	if status == "ACTIVE" {
+		var exists bool
+		if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM veyra_strategy_version WHERE id=$1 AND strategy_id=$2)`, versionID, id).Scan(&exists); err != nil {
+			return model.StrategyDetail{}, mapError(err)
+		}
+		if !exists {
+			return model.StrategyDetail{}, apperror.ErrNotFound
+		}
+		selected = &versionID
+	} else {
+		selected = s.ActiveVersionID
+	}
+	_, err = tx.Exec(ctx, `UPDATE veyra_strategy SET status=$2,active_version_id=$3,version=version+1,updated_at=now() WHERE id=$1`, id, status, selected)
+	if err != nil {
+		return model.StrategyDetail{}, mapError(err)
+	}
+	if err := appendAudit(ctx, tx, actor, "STRATEGY_DEPLOYMENT_CHANGED", "strategy", id, requestID, map[string]any{"status": status, "version_id": selected, "reason": reason}); err != nil {
+		return model.StrategyDetail{}, err
+	}
+	detail, err := strategyDetail(ctx, tx, id)
+	if err != nil {
+		return detail, err
+	}
+	return detail, mapError(tx.Commit(ctx))
+}

internal/repository/trade_fill.go

@@ -0,0 +1,334 @@
+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())
+}

internal/repository/trade_plan.go

@@ -0,0 +1,258 @@
+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"
+	"time"
+)
+
+type TradePlanRepository struct{ pool *pgxpool.Pool }
+
+func NewTradePlanRepository(pool *pgxpool.Pool) *TradePlanRepository {
+	return &TradePlanRepository{pool}
+}
+func lockSimulationAccount(ctx context.Context, tx pgx.Tx, user, id string) error {
+	var mode, status string
+	if err := tx.QueryRow(ctx, `SELECT mode,ledger_status FROM veyra_account WHERE id=$1 AND user_id=$2 FOR UPDATE`, id, user).Scan(&mode, &status); err != nil {
+		return mapError(err)
+	}
+	if mode != "simulation" || status != "COMPLETE" {
+		return apperror.ErrInvalidInput
+	}
+	return nil
+}
+
+const activePlanPredicate = `p.status IN ('ACTIVE','PART_FILLED') AND p.valid_until>now() AND r.status='PUBLISHED' AND c.published_revision_id=r.id AND r.valid_until>now()`
+
+func budgetMinor(ctx context.Context, tx pgx.Tx, account, exclude string) (int64, int64, error) {
+	var cash, reserved 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 0, 0, mapError(err)
+	}
+	if err := tx.QueryRow(ctx, `SELECT COALESCE(sum(p.reserved_minor),0)::bigint FROM veyra_trade_plan p JOIN veyra_candidate_revision r ON r.id=p.candidate_revision_id JOIN veyra_candidate c ON c.id=r.candidate_id WHERE p.account_id=$1 AND p.id<>$2 AND `+activePlanPredicate, account, exclude).Scan(&reserved); err != nil {
+		return 0, 0, mapError(err)
+	}
+	return cash, reserved, nil
+}
+func (r *TradePlanRepository) Preview(ctx context.Context, user, account, candidate string, version, qty, fee int64) (model.PlanPreview, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return model.PlanPreview{}, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	publication, err := lockCandidate(ctx, tx, candidate)
+	if err != nil {
+		return model.PlanPreview{}, err
+	}
+	if publication.Version != version || !publication.IsActivePublication {
+		return model.PlanPreview{}, apperror.ErrCandidateUnavailable
+	}
+	if err := lockSimulationAccount(ctx, tx, user, account); err != nil {
+		return model.PlanPreview{}, err
+	}
+	high, err := model.ParseMoney(publication.Snapshot.ReferencePlan.EntryHigh)
+	if err != nil {
+		return model.PlanPreview{}, err
+	}
+	needed := high*qty + fee
+	if needed <= 0 || needed > 100000000000 {
+		return model.PlanPreview{}, apperror.ErrInvalidInput
+	}
+	cash, reserved, err := budgetMinor(ctx, tx, account, "")
+	if err != nil {
+		return model.PlanPreview{}, err
+	}
+	if needed > cash-reserved {
+		return model.PlanPreview{}, apperror.ErrInsufficientCash
+	}
+	preview := model.PlanPreview{ID: uuid.NewString(), AccountID: account, CandidateRevisionID: candidate, CandidateVersion: version, Quantity: qty, FeeBudgetMinor: fee, ReservedMinor: needed, ReservedAmount: model.FormatMoney(needed), FeeBudget: model.FormatMoney(fee), ExpiresAt: time.Now().UTC().Add(5 * time.Minute)}
+	if publication.ValidUntil.Before(preview.ExpiresAt) {
+		preview.ExpiresAt = publication.ValidUntil
+	}
+	err = tx.QueryRow(ctx, `INSERT INTO veyra_plan_preview(id,account_id,candidate_revision_id,candidate_version,quantity,fee_budget_minor,reserved_minor,expires_at) VALUES($1,$2,$3,$4,$5,$6,$7,$8) RETURNING created_at`, preview.ID, account, candidate, version, qty, fee, needed, preview.ExpiresAt).Scan(&preview.CreatedAt)
+	if err != nil {
+		return preview, mapError(err)
+	}
+	return preview, mapError(tx.Commit(ctx))
+}
+
+const tradePlanColumns = `p.id,p.preview_id,p.account_id,p.candidate_revision_id,p.candidate_version,p.quantity,p.filled_quantity,p.fee_budget_minor,p.reserved_minor,p.status,p.version,p.valid_until,p.created_at,p.updated_at,r.snapshot`
+const tradePlanJoin = ` FROM veyra_trade_plan p JOIN veyra_candidate_revision r ON r.id=p.candidate_revision_id JOIN veyra_candidate c ON c.id=r.candidate_id JOIN veyra_account a ON a.id=p.account_id `
+
+func scanTradePlan(row pgx.Row) (model.TradePlan, error) {
+	var p model.TradePlan
+	var snapshot []byte
+	err := row.Scan(&p.ID, &p.PreviewID, &p.AccountID, &p.CandidateRevisionID, &p.CandidateVersion, &p.Quantity, &p.FilledQuantity, &p.FeeBudgetMinor, &p.ReservedMinor, &p.Status, &p.Version, &p.ValidUntil, &p.CreatedAt, &p.UpdatedAt, &snapshot)
+	if err != nil {
+		return p, mapError(err)
+	}
+	if json.Unmarshal(snapshot, &p.CandidateContent) != nil {
+		return p, apperror.ErrUnavailable
+	}
+	p.ReservedAmount = model.FormatMoney(p.ReservedMinor)
+	return p, nil
+}
+func (r *TradePlanRepository) Confirm(ctx context.Context, user, id, requestID string) (model.TradePlan, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return model.TradePlan{}, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	var preview model.PlanPreview
+	err = tx.QueryRow(ctx, `SELECT v.id,v.account_id,v.candidate_revision_id,v.candidate_version,v.quantity,v.fee_budget_minor,v.reserved_minor,v.expires_at FROM veyra_plan_preview v JOIN veyra_account a ON a.id=v.account_id WHERE v.id=$1 AND a.user_id=$2`, id, user).Scan(&preview.ID, &preview.AccountID, &preview.CandidateRevisionID, &preview.CandidateVersion, &preview.Quantity, &preview.FeeBudgetMinor, &preview.ReservedMinor, &preview.ExpiresAt)
+	if err != nil {
+		return model.TradePlan{}, mapError(err)
+	}
+	publication, err := lockCandidate(ctx, tx, preview.CandidateRevisionID)
+	if err != nil {
+		return model.TradePlan{}, err
+	}
+	if err := lockSimulationAccount(ctx, tx, user, preview.AccountID); err != nil {
+		return model.TradePlan{}, err
+	}
+	// Retrying a confirmed preview returns the same plan, even after its preview expires.
+	existing, err := scanTradePlan(tx.QueryRow(ctx, `SELECT `+tradePlanColumns+tradePlanJoin+` WHERE p.preview_id=$1 AND a.user_id=$2`, id, user))
+	if err == nil {
+		return existing, nil
+	}
+	if err != apperror.ErrNotFound {
+		return existing, err
+	}
+	if !preview.ExpiresAt.After(time.Now()) {
+		return model.TradePlan{}, apperror.ErrPreviewExpired
+	}
+	if !publication.IsActivePublication || publication.Version != preview.CandidateVersion {
+		return model.TradePlan{}, apperror.ErrCandidateUnavailable
+	}
+	cash, reserved, err := budgetMinor(ctx, tx, preview.AccountID, "")
+	if err != nil {
+		return model.TradePlan{}, err
+	}
+	if preview.ReservedMinor > cash-reserved {
+		return model.TradePlan{}, apperror.ErrInsufficientCash
+	}
+	planID := uuid.NewString()
+	_, err = tx.Exec(ctx, `INSERT INTO veyra_trade_plan(id,preview_id,account_id,candidate_revision_id,candidate_version,quantity,fee_budget_minor,reserved_minor,valid_until) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)`, planID, id, preview.AccountID, preview.CandidateRevisionID, preview.CandidateVersion, preview.Quantity, preview.FeeBudgetMinor, preview.ReservedMinor, publication.ValidUntil)
+	if err != nil {
+		return model.TradePlan{}, mapError(err)
+	}
+	if err := appendAudit(ctx, tx, user, "PLAN_CONFIRMED", "trade_plan", planID, requestID, map[string]any{"preview_id": id, "reserved_amount": model.FormatMoney(preview.ReservedMinor)}); err != nil {
+		return model.TradePlan{}, err
+	}
+	plan, err := scanTradePlan(tx.QueryRow(ctx, `SELECT `+tradePlanColumns+tradePlanJoin+` WHERE p.id=$1`, planID))
+	if err != nil {
+		return plan, err
+	}
+	return plan, mapError(tx.Commit(ctx))
+}
+func invalidatePlans(ctx context.Context, tx pgx.Tx, user string) error {
+	rows, err := tx.Query(ctx, `UPDATE veyra_trade_plan p SET status='INVALIDATED',reserved_minor=0,version=p.version+1,updated_at=now() FROM veyra_candidate_revision r,veyra_candidate c,veyra_account a WHERE p.candidate_revision_id=r.id AND c.id=r.candidate_id AND a.id=p.account_id AND a.user_id=$1 AND p.status IN ('ACTIVE','PART_FILLED') AND (p.valid_until<=now() OR r.valid_until<=now() OR r.status<>'PUBLISHED' OR c.published_revision_id IS DISTINCT FROM r.id) RETURNING p.id,p.version`, user)
+	if err != nil {
+		return mapError(err)
+	}
+	type invalid struct {
+		id      string
+		version int64
+	}
+	list := make([]invalid, 0)
+	for rows.Next() {
+		var v invalid
+		if err := rows.Scan(&v.id, &v.version); err != nil {
+			rows.Close()
+			return mapError(err)
+		}
+		list = append(list, v)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return mapError(err)
+	}
+	for _, v := range list {
+		eventID := uuid.NewString()
+		if err := appendAudit(ctx, tx, user, "PLAN_INVALIDATED", "trade_plan", v.id, eventID, map[string]any{"cause": "candidate_unavailable_or_expired", "version": v.version}); err != nil {
+			return err
+		}
+		_, err = tx.Exec(ctx, `INSERT INTO veyra_outbox_event(id,aggregate_type,aggregate_id,aggregate_version,event_type,payload) VALUES($1,'trade_plan',$2,$3,'plan.invalidated','{}'::jsonb)`, eventID, v.id, v.version)
+		if err != nil {
+			return mapError(err)
+		}
+	}
+	return nil
+}
+func (r *TradePlanRepository) List(ctx context.Context, user string) ([]model.TradePlan, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return nil, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	if err := invalidatePlans(ctx, tx, user); err != nil {
+		return nil, err
+	}
+	rows, err := tx.Query(ctx, `SELECT `+tradePlanColumns+tradePlanJoin+` WHERE a.user_id=$1 ORDER BY p.created_at DESC,p.id DESC LIMIT 100`, user)
+	if err != nil {
+		return nil, mapError(err)
+	}
+	list := make([]model.TradePlan, 0)
+	for rows.Next() {
+		p, err := scanTradePlan(rows)
+		if err != nil {
+			rows.Close()
+			return nil, err
+		}
+		list = append(list, p)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return nil, mapError(err)
+	}
+	return list, mapError(tx.Commit(ctx))
+}
+func (r *TradePlanRepository) Cancel(ctx context.Context, user, id string, version int64, requestID string) (model.TradePlan, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return model.TradePlan{}, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	var account string
+	if err := tx.QueryRow(ctx, `SELECT p.account_id FROM veyra_trade_plan p JOIN veyra_account a ON a.id=p.account_id WHERE p.id=$1 AND a.user_id=$2`, id, user).Scan(&account); err != nil {
+		return model.TradePlan{}, mapError(err)
+	}
+	if err := lockSimulationAccount(ctx, tx, user, account); err != nil {
+		return model.TradePlan{}, err
+	}
+	tag, err := tx.Exec(ctx, `UPDATE veyra_trade_plan SET status='CANCELLED',reserved_minor=0,version=version+1,updated_at=now() WHERE id=$1 AND version=$2 AND status IN ('ACTIVE','PART_FILLED')`, id, version)
+	if err != nil {
+		return model.TradePlan{}, mapError(err)
+	}
+	if tag.RowsAffected() != 1 {
+		return model.TradePlan{}, apperror.ErrConflict
+	}
+	if err := appendAudit(ctx, tx, user, "PLAN_CANCELLED", "trade_plan", id, requestID, map[string]any{"input_version": version}); err != nil {
+		return model.TradePlan{}, err
+	}
+	plan, err := scanTradePlan(tx.QueryRow(ctx, `SELECT `+tradePlanColumns+tradePlanJoin+` WHERE p.id=$1`, id))
+	if err != nil {
+		return plan, err
+	}
+	return plan, mapError(tx.Commit(ctx))
+}
+func (r *TradePlanRepository) Budget(ctx context.Context, user, account string) (model.PlanBudget, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return model.PlanBudget{}, mapError(err)
+	}
+	defer tx.Rollback(context.Background())
+	if err := lockSimulationAccount(ctx, tx, user, account); err != nil {
+		return model.PlanBudget{}, err
+	}
+	cash, reserved, err := budgetMinor(ctx, tx, account, "")
+	if err != nil {
+		return model.PlanBudget{}, err
+	}
+	return model.PlanBudget{AccountID: account, CashBalance: model.FormatMoney(cash), ReservedAmount: model.FormatMoney(reserved), AvailableAmount: model.FormatMoney(cash - reserved)}, nil
+}