314 lines
13 KiB
Go
314 lines
13 KiB
Go
package integration
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"go.uber.org/zap"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/model"
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/repository"
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/service"
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
func testScreeningLifecycle(t *testing.T, ctx context.Context, pool *pgxpool.Pool, admin, user model.User) {
|
|
strategies := service.NewStrategyService(repository.NewStrategyRepository(pool), true)
|
|
input := service.StrategyInput{Code: "auto_screening", Name: "自动筛选测试", Description: "integration fixture", Parameters: model.StrategyParameters{ShortWindow: 2, LongWindow: 3, LotSize: 100, Screening: &model.ScreeningParameters{Securities: []model.ScreeningSecurity{{Symbol: "600000", Exchange: "SSE"}}, EntryBandBPS: 100, StopLossBPS: 500, TargetBPS: 1000, ValidHours: 24}}}
|
|
st, err := strategies.Save(ctx, admin, "", input, uuid.NewString())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
st, err = strategies.Deploy(ctx, admin, st.ID, st.Versions[0].ID, "ACTIVE", st.Version, "fixture activation", uuid.NewString())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
providerRepo := repository.NewProviderRepository(pool)
|
|
provider, err := providerRepo.Get(ctx, "market_primary")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// Only disposable-schema fixtures are modified, never the development configuration.
|
|
if _, err := pool.Exec(ctx, `UPDATE veyra_provider_config SET enabled=true,market_adapter='sina_public',history_base_url='https://quotes.example.cn' WHERE id=$1`, provider.ID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
provider, err = providerRepo.Get(ctx, "market_primary")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
observed := time.Now().UTC().Add(-23 * time.Hour)
|
|
instrument := model.Instrument{Symbol: "600000", Exchange: "SSE", Name: "fixture", PriceCents: 2001, ObservedAt: observed}
|
|
if err := repository.NewMarketRepository(pool).Save(ctx, provider, []model.Instrument{instrument}, time.Now().UTC(), ""); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
searchRepo := repository.NewMarketRepository(pool)
|
|
if err := searchRepo.Save(ctx, provider, []model.Instrument{{Symbol: "002215", Exchange: "SZSE", Name: "诺 普 信", PriceCents: 1000, ObservedAt: observed}}, time.Now().UTC(), ""); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, query := range []string{"诺普信", "诺 普 信", " 诺\t普 信 ", "002215", "002 215"} {
|
|
stocks, err := searchRepo.List(ctx, query)
|
|
if err != nil || len(stocks) != 1 || stocks[0].Symbol != "002215" || stocks[0].Name != "诺 普 信" {
|
|
t.Fatalf("padded security search %q: %v, %v", query, stocks, err)
|
|
}
|
|
}
|
|
for _, query := range []string{"not-a-security", "%", "_"} {
|
|
stocks, err := searchRepo.List(ctx, query)
|
|
if err != nil || len(stocks) != 0 {
|
|
t.Fatalf("literal security search %q matched: %v, %v", query, stocks, err)
|
|
}
|
|
}
|
|
repo := repository.NewScreeningRepository(pool)
|
|
targets, err := repo.Targets(ctx, provider)
|
|
if err != nil || len(targets) != 1 {
|
|
t.Fatalf("targets %d: %v", len(targets), err)
|
|
}
|
|
target := targets[0]
|
|
zone := time.FixedZone("Shanghai", 8*3600)
|
|
run := model.StrategyScreening{StrategyVersionID: target.Version.ID, Symbol: "600000", Exchange: "SSE", BarDate: observed.In(zone).Format("2006-01-02"), SourceID: provider.ID, SourceVersion: provider.Version, SourceAdapter: "sina_public", ObservedAt: observed, ReceivedAt: time.Now().UTC(), InputHash: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", Input: json.RawMessage(`{"fixture":true}`), Status: "SIGNAL"}
|
|
draft := model.CandidateDraftInput{ObservedAt: observed, ValidUntil: time.Now().UTC().Add(time.Hour), Content: model.CandidateContent{Symbol: "600000", Exchange: "SSE", Name: "fixture", StrategyCode: "trend_following", Title: "fixture signal", Thesis: "fixture", CounterEvidence: "fixture risk", Invalidation: "fixture pause", Evidence: []model.CandidateEvidence{{SourceURL: "https://quotes.example.cn", Locator: "fixture", Summary: "fixture"}}, ReferencePlan: model.CandidateReferencePlan{EntryLow: "19.80", EntryHigh: "20.22", StopLoss: "19.00", TargetPrice: "22.02"}}}
|
|
var before int
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM veyra_candidate`).Scan(&before); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var wg sync.WaitGroup
|
|
results := make(chan error, 2)
|
|
for range 2 {
|
|
wg.Add(1)
|
|
go func() { defer wg.Done(); results <- repo.Record(ctx, target, run, &draft) }()
|
|
}
|
|
wg.Wait()
|
|
close(results)
|
|
success, conflict := 0, 0
|
|
for e := range results {
|
|
if e == nil {
|
|
success++
|
|
} else if errors.Is(e, apperror.ErrConflict) {
|
|
conflict++
|
|
} else {
|
|
t.Fatal(e)
|
|
}
|
|
}
|
|
if success != 1 || conflict != 1 {
|
|
t.Fatal("concurrent duplicate screening did not converge")
|
|
}
|
|
var after int
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM veyra_candidate`).Scan(&after); err != nil || after != before+1 {
|
|
t.Fatal("duplicate screening left orphan candidate")
|
|
}
|
|
records, err := repo.List(ctx)
|
|
if err != nil || len(records) != 1 || records[0].CandidateRevisionID == nil {
|
|
t.Fatal("screening provenance missing")
|
|
}
|
|
if _, err := pool.Exec(ctx, `UPDATE veyra_strategy_screening SET status='NO_SIGNAL' WHERE id=$1`, records[0].ID); err == nil {
|
|
t.Fatal("screening evidence mutable")
|
|
}
|
|
recordDetail, err := repo.Detail(ctx, records[0].ID)
|
|
if err != nil || string(recordDetail.Input) != "{\"fixture\": true}" {
|
|
t.Fatalf("immutable input detail: %s %v", recordDetail.Input, err)
|
|
}
|
|
candidates := service.NewCandidateService(repository.NewCandidateRepository(pool), true)
|
|
candidate, err := candidates.Detail(ctx, admin, *records[0].CandidateRevisionID, true)
|
|
if err != nil || candidate.Status != "PENDING_REVIEW" || candidate.Screening == nil {
|
|
t.Fatalf("review queue provenance: %v", err)
|
|
}
|
|
if _, err := candidates.Detail(ctx, user, candidate.ID, false); !errors.Is(err, apperror.ErrNotFound) {
|
|
t.Fatal("unreviewed signal leaked to App")
|
|
}
|
|
pub, err := candidates.Transition(ctx, admin, candidate.ID, "PUBLISH", uuid.NewString(), model.CandidateActionInput{ExpectedVersion: candidate.Version, Reason: "fixture review"})
|
|
if err != nil || !pub.IsActivePublication {
|
|
t.Fatalf("review publish: %v", err)
|
|
}
|
|
accounts := service.NewAccountService(repository.NewAccountRepository(pool))
|
|
cash := "100000.00"
|
|
auth, err := service.NewAuthService(repository.NewIdentityRepository(pool))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
owner, err := auth.ProvisionUser(ctx, "scanner_user", "a-test-only-password", "Scanner User", "USER")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
account, err := accounts.Create(ctx, owner.ID, "simulation", "scanner fixture", &cash)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
plans := service.NewTradePlanService(repository.NewTradePlanRepository(pool), true)
|
|
preview, err := plans.Preview(ctx, owner, service.PlanPreviewInput{AccountID: account.ID, CandidateRevisionID: pub.ID, ExpectedCandidateVersion: pub.Version, Quantity: "100", FeeBudget: "0"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
plan, err := plans.Confirm(ctx, owner, preview.ID, uuid.NewString())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// Real public quote ingestion uses the existing outbox and never writes ledger facts.
|
|
if err := repository.NewMarketRepository(pool).Save(ctx, provider, []model.Instrument{instrument}, time.Now().UTC(), ""); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var publicCount int
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM veyra_quote_snapshot WHERE data_mode='public'`).Scan(&publicCount); err != nil || publicCount != 0 {
|
|
t.Fatal("old market close queued a public alert")
|
|
}
|
|
instrument.ObservedAt = time.Now().UTC()
|
|
marketRepo := repository.NewMarketRepository(pool)
|
|
if err := marketRepo.Save(ctx, provider, []model.Instrument{instrument}, time.Now().UTC(), ""); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
notifications := repository.NewNotificationRepository(pool)
|
|
for range 100 {
|
|
worked, err := notifications.ProcessNext(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !worked {
|
|
break
|
|
}
|
|
}
|
|
notices, err := notifications.List(ctx, owner.ID)
|
|
if err != nil || len(notices) != 1 {
|
|
t.Fatalf("public alert missing: %v count %d", err, len(notices))
|
|
}
|
|
if !strings.Contains(notices[0].Body, "公开行情价格") {
|
|
t.Fatal("public price mislabeled as manual simulation")
|
|
}
|
|
if err := marketRepo.Save(ctx, provider, []model.Instrument{instrument}, time.Now().UTC(), ""); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for range 100 {
|
|
worked, err := notifications.ProcessNext(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !worked {
|
|
break
|
|
}
|
|
}
|
|
notices, err = notifications.List(ctx, owner.ID)
|
|
if err != nil || len(notices) != 1 {
|
|
t.Fatal("unchanged quote duplicated reminder")
|
|
}
|
|
var balance int64
|
|
var fills int
|
|
if err := pool.QueryRow(ctx, `SELECT sum(amount_minor)::bigint FROM veyra_cash_entry WHERE account_id=$1`, account.ID).Scan(&balance); err != nil || balance != 10000000 {
|
|
t.Fatal("public alert altered cash")
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM veyra_trade_fill WHERE account_id=$1`, account.ID).Scan(&fills); err != nil || fills != 0 {
|
|
t.Fatal("public alert posted trade")
|
|
}
|
|
// A queued observation from a disabled source must not trigger another plan.
|
|
preview2, err := plans.Preview(ctx, owner, service.PlanPreviewInput{AccountID: account.ID, CandidateRevisionID: pub.ID, ExpectedCandidateVersion: pub.Version, Quantity: "100", FeeBudget: "0"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := plans.Confirm(ctx, owner, preview2.ID, uuid.NewString()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
instrument.ObservedAt = time.Now().UTC()
|
|
if err := marketRepo.Save(ctx, provider, []model.Instrument{instrument}, time.Now().UTC(), ""); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := pool.Exec(ctx, `UPDATE veyra_provider_config SET enabled=false WHERE id=$1`, provider.ID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for range 100 {
|
|
worked, err := notifications.ProcessNext(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !worked {
|
|
break
|
|
}
|
|
}
|
|
notices, err = notifications.List(ctx, owner.ID)
|
|
if err != nil || len(notices) != 1 {
|
|
t.Fatal("disabled source generated delayed reminder")
|
|
}
|
|
if _, err := pool.Exec(ctx, `UPDATE veyra_provider_config SET enabled=true WHERE id=$1`, provider.ID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
st, err = strategies.Deploy(ctx, admin, st.ID, "", "PAUSED", st.Version, "fixture pause", uuid.NewString())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
detail, err := candidates.Detail(ctx, user, pub.ID, false)
|
|
if err != nil || detail.IsActivePublication {
|
|
t.Fatal("paused signal stayed actionable")
|
|
}
|
|
if _, err := plans.Preview(ctx, owner, service.PlanPreviewInput{AccountID: account.ID, CandidateRevisionID: pub.ID, ExpectedCandidateVersion: pub.Version, Quantity: "100", FeeBudget: "0"}); !errors.Is(err, apperror.ErrCandidateUnavailable) {
|
|
t.Fatal("paused strategy accepted plan")
|
|
}
|
|
list, err := plans.List(ctx, owner)
|
|
if err != nil || len(list) != 2 {
|
|
t.Fatalf("pause did not release plan: %v", err)
|
|
}
|
|
found := false
|
|
for _, p := range list {
|
|
if p.ID == plan.ID {
|
|
found = true
|
|
}
|
|
if p.Status != "INVALIDATED" || p.ReservedMinor != 0 {
|
|
t.Fatal("paused plan kept reserve")
|
|
}
|
|
}
|
|
if !found {
|
|
t.Fatal("original plan disappeared")
|
|
}
|
|
run.Symbol = "600001"
|
|
if err := repo.Record(ctx, target, run, &draft); !errors.Is(err, apperror.ErrConflict) {
|
|
t.Fatal("paused deployment accepted late worker")
|
|
}
|
|
worker := service.NewScreeningService(repo, service.NewMarketService(repository.NewMarketRepository(pool), providerRepo, idleScreeningAdapter{}, true), true)
|
|
if _, err := worker.RequestRun(user); !errors.Is(err, apperror.ErrForbidden) {
|
|
t.Fatal("app user requested research run")
|
|
}
|
|
first, err := worker.RequestRun(admin)
|
|
if err != nil || first.State != "QUEUED" {
|
|
t.Fatal("run did not queue")
|
|
}
|
|
second, err := worker.RequestRun(admin)
|
|
if err != nil || second.State != "QUEUED" {
|
|
t.Fatal("duplicate queue request changed state")
|
|
}
|
|
workerCtx, cancel := context.WithCancel(ctx)
|
|
done := make(chan struct{})
|
|
go func() { defer close(done); worker.Run(workerCtx, zap.NewNop()) }()
|
|
deadline := time.After(3 * time.Second)
|
|
for {
|
|
state, err := worker.State(admin)
|
|
if err != nil {
|
|
cancel()
|
|
<-done
|
|
t.Fatal(err)
|
|
}
|
|
if state.State == "COMPLETED" && state.FinishedAt != nil {
|
|
break
|
|
}
|
|
select {
|
|
case <-deadline:
|
|
cancel()
|
|
<-done
|
|
t.Fatal("queued worker did not complete")
|
|
case <-time.After(10 * time.Millisecond):
|
|
}
|
|
}
|
|
cancel()
|
|
<-done
|
|
|
|
}
|
|
|
|
// There are no eligible targets after pause, so this adapter must never be called.
|
|
type idleScreeningAdapter struct{}
|
|
|
|
func (idleScreeningAdapter) Snapshot(context.Context, model.ProviderConfig) ([]model.Instrument, error) {
|
|
return nil, apperror.ErrUnavailable
|
|
}
|
|
func (idleScreeningAdapter) History(context.Context, model.ProviderConfig, string, string, string, string) (model.MarketHistory, error) {
|
|
return model.MarketHistory{}, apperror.ErrUnavailable
|
|
}
|