129 lines
7.4 KiB
Go
129 lines
7.4 KiB
Go
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/pgxpool"
|
|
)
|
|
|
|
type ScreeningRepository struct{ pool *pgxpool.Pool }
|
|
|
|
func NewScreeningRepository(pool *pgxpool.Pool) *ScreeningRepository {
|
|
return &ScreeningRepository{pool}
|
|
}
|
|
func (r *ScreeningRepository) Targets(ctx context.Context, c model.ProviderConfig) ([]model.ScreeningTarget, error) {
|
|
rows, err := r.pool.Query(ctx, `SELECT s.id,s.name,v.id,v.strategy_id,v.revision,v.algorithm,v.parameters,v.description,v.author_id,v.created_at,i.id,i.symbol,i.exchange,i.name,i.price_cents,i.observed_at,i.received_at,i.source_id,i.source_version
|
|
FROM veyra_strategy s JOIN veyra_strategy_version v ON v.id=s.active_version_id
|
|
CROSS JOIN LATERAL jsonb_array_elements(COALESCE(v.parameters->'screening'->'securities','[]'::jsonb)) security
|
|
JOIN veyra_instrument i ON i.symbol=security->>'symbol' AND i.exchange=security->>'exchange'
|
|
JOIN veyra_user u ON u.id=v.author_id AND u.role='ADMIN' AND u.status='ACTIVE'
|
|
WHERE s.status='ACTIVE' AND i.source_id=$1 AND i.source_version=$2 AND i.observed_at>=now()-interval '24 hours' AND i.observed_at<=now()+interval '1 minute' AND i.received_at>=now()-interval '10 minutes'
|
|
AND ((i.observed_at AT TIME ZONE 'Asia/Shanghai')::date < (now() AT TIME ZONE 'Asia/Shanghai')::date OR EXTRACT(hour FROM now() AT TIME ZONE 'Asia/Shanghai')>=16)
|
|
AND NOT EXISTS(SELECT 1 FROM veyra_strategy_screening x WHERE x.strategy_version_id=v.id AND x.symbol=i.symbol AND x.exchange=i.exchange AND x.bar_date=(i.observed_at AT TIME ZONE 'Asia/Shanghai')::date AND x.source_id=$1 AND x.source_version=$2)
|
|
ORDER BY s.updated_at,s.id,i.symbol LIMIT 2000`, c.ID, c.Version)
|
|
if err != nil {
|
|
return nil, mapError(err)
|
|
}
|
|
defer rows.Close()
|
|
targets := make([]model.ScreeningTarget, 0)
|
|
for rows.Next() {
|
|
var t model.ScreeningTarget
|
|
var raw []byte
|
|
v := &t.Version
|
|
i := &t.Instrument
|
|
if err := rows.Scan(&t.StrategyID, &t.StrategyName, &v.ID, &v.StrategyID, &v.Revision, &v.Algorithm, &raw, &v.Description, &v.AuthorID, &v.CreatedAt, &i.ID, &i.Symbol, &i.Exchange, &i.Name, &i.PriceCents, &i.ObservedAt, &i.ReceivedAt, &i.SourceID, &i.SourceVersion); err != nil {
|
|
return nil, mapError(err)
|
|
}
|
|
if json.Unmarshal(raw, &v.Parameters) != nil {
|
|
return nil, apperror.ErrUnavailable
|
|
}
|
|
targets = append(targets, t)
|
|
}
|
|
return targets, mapError(rows.Err())
|
|
}
|
|
func (r *ScreeningRepository) Record(ctx context.Context, t model.ScreeningTarget, run model.StrategyScreening, draft *model.CandidateDraftInput) error {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return mapError(err)
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
// Lock deployment before creating a candidate; pause/version changes cannot race publication.
|
|
var status string
|
|
var active *string
|
|
if err := tx.QueryRow(ctx, `SELECT status,active_version_id FROM veyra_strategy WHERE id=$1 FOR SHARE`, t.StrategyID).Scan(&status, &active); err != nil {
|
|
return mapError(err)
|
|
}
|
|
if status != "ACTIVE" || active == nil || *active != t.Version.ID {
|
|
return apperror.ErrConflict
|
|
}
|
|
var sourceVersion int64
|
|
var enabled bool
|
|
if err := tx.QueryRow(ctx, `SELECT version,enabled FROM veyra_provider_config WHERE id=$1 FOR SHARE`, run.SourceID).Scan(&sourceVersion, &enabled); err != nil {
|
|
return mapError(err)
|
|
}
|
|
if !enabled || sourceVersion != run.SourceVersion {
|
|
return apperror.ErrConflict
|
|
}
|
|
var actorValid bool
|
|
if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM veyra_user WHERE id=$1 AND status='ACTIVE' AND role='ADMIN')`, t.Version.AuthorID).Scan(&actorValid); err != nil {
|
|
return mapError(err)
|
|
}
|
|
if !actorValid {
|
|
return apperror.ErrConflict
|
|
}
|
|
requestID := uuid.NewString()
|
|
var revisionID *string
|
|
if draft != nil {
|
|
if err := model.ValidateDraft(*draft, time.Now().UTC()); err != nil {
|
|
return err
|
|
}
|
|
candidate, err := createCandidate(ctx, tx, *draft, t.Version.AuthorID, requestID, "", 0)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Machine submission leaves the approval decision to an administrator.
|
|
if _, err := tx.Exec(ctx, `UPDATE veyra_candidate_revision SET status='PENDING_REVIEW',version=version+1 WHERE id=$1`, candidate.ID); err != nil {
|
|
return mapError(err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO veyra_candidate_review(id,revision_id,reviewer_id,decision,input_version,reason) VALUES($1,$2,$3,'SUBMIT',1,'自动日线筛选提交，等待人工审核')`, uuid.NewString(), candidate.ID, t.Version.AuthorID); err != nil {
|
|
return mapError(err)
|
|
}
|
|
revisionID = &candidate.ID
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO veyra_strategy_screening(id,strategy_version_id,symbol,exchange,bar_date,source_id,source_version,source_adapter,observed_at,received_at,input_hash,input,status,candidate_revision_id) VALUES($1,$2,$3,$4,$5::date,$6,$7,$8,$9,$10,$11,$12,$13,$14)`, uuid.NewString(), t.Version.ID, run.Symbol, run.Exchange, run.BarDate, run.SourceID, run.SourceVersion, run.SourceAdapter, run.ObservedAt, run.ReceivedAt, run.InputHash, run.Input, run.Status, revisionID)
|
|
if err != nil {
|
|
return mapError(err)
|
|
} // Unique key rolls back candidate/audit too, including concurrent workers.
|
|
if err := appendAudit(ctx, tx, t.Version.AuthorID, "STRATEGY_SCREENING_COMPLETED", "strategy_version", t.Version.ID, requestID, map[string]any{"symbol": run.Symbol, "exchange": run.Exchange, "bar_date": run.BarDate, "status": run.Status, "input_hash": run.InputHash, "candidate_revision_id": revisionID}); err != nil {
|
|
return err
|
|
}
|
|
return mapError(tx.Commit(ctx))
|
|
}
|
|
func (r *ScreeningRepository) List(ctx context.Context) ([]model.StrategyScreening, error) {
|
|
rows, err := r.pool.Query(ctx, `SELECT x.id,x.strategy_version_id,s.name,v.revision,x.symbol,x.exchange,x.bar_date::text,x.source_id,x.source_version,x.source_adapter,x.observed_at,x.received_at,x.input_hash,x.status,x.candidate_revision_id,x.created_at FROM veyra_strategy_screening x JOIN veyra_strategy_version v ON v.id=x.strategy_version_id JOIN veyra_strategy s ON s.id=v.strategy_id ORDER BY x.created_at DESC,x.id DESC LIMIT 100`)
|
|
if err != nil {
|
|
return nil, mapError(err)
|
|
}
|
|
defer rows.Close()
|
|
list := make([]model.StrategyScreening, 0)
|
|
for rows.Next() {
|
|
var x model.StrategyScreening
|
|
if err := rows.Scan(&x.ID, &x.StrategyVersionID, &x.StrategyName, &x.StrategyRevision, &x.Symbol, &x.Exchange, &x.BarDate, &x.SourceID, &x.SourceVersion, &x.SourceAdapter, &x.ObservedAt, &x.ReceivedAt, &x.InputHash, &x.Status, &x.CandidateRevisionID, &x.CreatedAt); err != nil {
|
|
return nil, mapError(err)
|
|
}
|
|
list = append(list, x)
|
|
}
|
|
return list, mapError(rows.Err())
|
|
}
|
|
|
|
func (r *ScreeningRepository) Detail(ctx context.Context, id string) (model.StrategyScreening, error) {
|
|
var x model.StrategyScreening
|
|
err := r.pool.QueryRow(ctx, `SELECT x.id,x.strategy_version_id,s.name,v.revision,x.symbol,x.exchange,x.bar_date::text,x.source_id,x.source_version,x.source_adapter,x.observed_at,x.received_at,x.input_hash,x.status,x.candidate_revision_id,x.created_at,x.input FROM veyra_strategy_screening x JOIN veyra_strategy_version v ON v.id=x.strategy_version_id JOIN veyra_strategy s ON s.id=v.strategy_id WHERE x.id=$1`, id).Scan(&x.ID, &x.StrategyVersionID, &x.StrategyName, &x.StrategyRevision, &x.Symbol, &x.Exchange, &x.BarDate, &x.SourceID, &x.SourceVersion, &x.SourceAdapter, &x.ObservedAt, &x.ReceivedAt, &x.InputHash, &x.Status, &x.CandidateRevisionID, &x.CreatedAt, &x.Input)
|
|
return x, mapError(err)
|
|
}
|