package service import ( "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "sync" "time" "git.sundynix.cn/Blizzard/Veyra/internal/apperror" "git.sundynix.cn/Blizzard/Veyra/internal/model" "github.com/google/uuid" "go.uber.org/zap" ) type ScreeningRepository interface { Targets(context.Context, model.ProviderConfig) ([]model.ScreeningTarget, error) Record(context.Context, model.ScreeningTarget, model.StrategyScreening, *model.CandidateDraftInput) error List(context.Context) ([]model.StrategyScreening, error) Detail(context.Context, string) (model.StrategyScreening, error) } type ScreeningService struct { repo ScreeningRepository market *MarketService development bool cursor int mu sync.Mutex wake chan struct{} state model.ScreeningWorkerState } func NewScreeningService(repo ScreeningRepository, market *MarketService, development bool) *ScreeningService { return &ScreeningService{repo: repo, market: market, development: development, wake: make(chan struct{}, 1), state: model.ScreeningWorkerState{State: "IDLE", Message: "尚未运行"}} } func (s *ScreeningService) List(ctx context.Context, actor model.User) ([]model.StrategyScreening, error) { if actor.Role != "ADMIN" || actor.Status != "ACTIVE" { return nil, apperror.ErrForbidden } if s == nil || s.repo == nil || !s.development { return nil, apperror.ErrUnavailable } return s.repo.List(ctx) } func (s *ScreeningService) Detail(ctx context.Context, actor model.User, id string) (model.StrategyScreening, error) { if actor.Role != "ADMIN" || actor.Status != "ACTIVE" { return model.StrategyScreening{}, apperror.ErrForbidden } if s == nil || s.repo == nil || !s.development { return model.StrategyScreening{}, apperror.ErrUnavailable } if _, err := uuid.Parse(id); err != nil { return model.StrategyScreening{}, apperror.ErrInvalidInput } return s.repo.Detail(ctx, id) } func (s *ScreeningService) State(actor model.User) (model.ScreeningWorkerState, error) { if actor.Role != "ADMIN" || actor.Status != "ACTIVE" { return model.ScreeningWorkerState{}, apperror.ErrForbidden } if s == nil || s.repo == nil || s.market == nil || s.market.providers == nil || s.market.adapter == nil || !s.development { return model.ScreeningWorkerState{}, apperror.ErrUnavailable } s.mu.Lock() defer s.mu.Unlock() return s.state, nil } func (s *ScreeningService) RequestRun(actor model.User) (model.ScreeningWorkerState, error) { if _, err := s.State(actor); err != nil { return model.ScreeningWorkerState{}, err } s.mu.Lock() defer s.mu.Unlock() if s.state.State == "RUNNING" || s.state.State == "QUEUED" { return s.state, nil } s.state = model.ScreeningWorkerState{State: "QUEUED", Message: "已排队,等待后台开始筛选"} select { case s.wake <- struct{}{}: default: } return s.state, nil } var screeningZone = time.FixedZone("Asia/Shanghai", 8*60*60) // EvaluateScreening uses completed daily bars only. Re-fetching a historical quote // never makes it fresh, and an unclosed session cannot generate a daily signal. func EvaluateScreening(t model.ScreeningTarget, c model.ProviderConfig, h model.MarketHistory, now time.Time) (model.StrategyScreening, *model.CandidateDraftInput, error) { var result model.StrategyScreening p := t.Version.Parameters q := p.Screening i := t.Instrument if q == nil || model.ValidateScreening(q) != nil || p.ShortWindow < 2 || p.LongWindow <= p.ShortWindow || p.LongWindow > 200 || h.Symbol != i.Symbol || h.Exchange != i.Exchange || h.Source != c.MarketAdapter || h.SourceVersion != c.Version || h.Adjustment != "unadjusted" || i.SourceID != c.ID || i.SourceVersion != c.Version || i.ObservedAt.Before(now.Add(-24*time.Hour)) || i.ObservedAt.After(now.Add(time.Minute)) || i.ReceivedAt.Before(now.Add(-10*time.Minute)) || h.ReceivedAt.Before(now.Add(-10*time.Minute)) || h.ReceivedAt.After(now.Add(time.Minute)) { return result, nil, apperror.ErrUnavailable } day := i.ObservedAt.In(screeningZone).Format("2006-01-02") localNow := now.In(screeningZone) if day == localNow.Format("2006-01-02") && localNow.Hour() < 16 { return result, nil, apperror.ErrUnavailable } if len(h.Bars) < p.LongWindow+1 || h.Bars[len(h.Bars)-1].Date != day { return result, nil, apperror.ErrUnavailable } bars := append([]model.HistoricalBar(nil), h.Bars[len(h.Bars)-p.LongWindow-1:]...) closes := make([]int64, len(bars)) previous := "" for j, b := range bars { d, e := time.Parse("2006-01-02", b.Date) o, e1 := model.ParseMoney(b.Open) hi, e2 := model.ParseMoney(b.High) lo, e3 := model.ParseMoney(b.Low) cl, e4 := model.ParseMoney(b.Close) if e != nil || d.Format("2006-01-02") != b.Date || b.Date <= previous || b.Date > day || e1 != nil || e2 != nil || e3 != nil || e4 != nil || lo <= 0 || hi < lo || o < lo || o > hi || cl < lo || cl > hi { return result, nil, apperror.ErrInvalidInput } closes[j] = cl previous = b.Date } sum := func(end, n int) int64 { var v int64 for _, x := range closes[end-n : end] { v += x } return v } end := len(closes) prevShort, prevLong := sum(end-1, p.ShortWindow), sum(end-1, p.LongWindow) short, long := sum(end, p.ShortWindow), sum(end, p.LongWindow) signal := prevShort*int64(p.LongWindow) <= prevLong*int64(p.ShortWindow) && short*int64(p.LongWindow) > long*int64(p.ShortWindow) input, _ := json.Marshal(struct { Parameters model.StrategyParameters `json:"parameters"` Bars []model.HistoricalBar `json:"bars"` }{p, bars}) hash := sha256.Sum256(input) result = model.StrategyScreening{StrategyVersionID: t.Version.ID, Symbol: i.Symbol, Exchange: i.Exchange, BarDate: day, SourceID: c.ID, SourceVersion: c.Version, SourceAdapter: c.MarketAdapter, ObservedAt: i.ObservedAt, ReceivedAt: h.ReceivedAt, InputHash: hex.EncodeToString(hash[:]), Input: input, Status: "NO_SIGNAL"} if !signal { return result, nil, nil } base := closes[end-1] low := base * (10000 - q.EntryBandBPS) / 10000 high := (base*(10000+q.EntryBandBPS) + 9999) / 10000 stop := base * (10000 - q.StopLossBPS) / 10000 target := (base*(10000+q.TargetBPS) + 9999) / 10000 draft := model.CandidateDraftInput{ObservedAt: i.ObservedAt, ValidUntil: now.Add(time.Duration(q.ValidHours) * time.Hour), Content: model.CandidateContent{ Symbol: i.Symbol, Exchange: i.Exchange, Name: i.Name, StrategyCode: "trend_following", Title: i.Name + " · 日线均线交叉待审核", Thesis: fmt.Sprintf("已完成日线 %s 出现 %d / %d 日均线上穿。策略 %s 修订 %d,仅为规则筛选结果,需人工核验后发布。", day, p.ShortWindow, p.LongWindow, t.StrategyName, t.Version.Revision), Evidence: []model.CandidateEvidence{{SourceURL: c.HistoryBaseURL, Locator: fmt.Sprintf("%s:%s 日线 %s", i.Exchange, i.Symbol, day), Summary: fmt.Sprintf("来源 %s 配置版本 %d;未复权收盘价 %s;输入摘要 %s。参数与日线快照在筛选记录中保留。", c.MarketAdapter, c.Version, model.FormatMoney(base), result.InputHash)}}, CounterEvidence: "公开数据可能缺失或延迟;未复权均线会受除权影响。此规则未核验公告、停复牌、涨跌停、流动性及策略有效性。", Invalidation: "策略暂停或切换版本、候选过期或撤回后失效;复核价格区间及数据来源后方可发布。", ReferencePlan: model.CandidateReferencePlan{EntryLow: model.FormatMoney(low), EntryHigh: model.FormatMoney(high), StopLoss: model.FormatMoney(stop), TargetPrice: model.FormatMoney(target)}, }} if e := model.ValidateDraft(draft, now); e != nil { return result, nil, e } result.Status = "SIGNAL" return result, &draft, nil } func (s *ScreeningService) Process(ctx context.Context, logger *zap.Logger) (processErr error) { started := time.Now().UTC() s.mu.Lock() if s.state.State == "RUNNING" { s.mu.Unlock() return apperror.ErrConflict } s.state = model.ScreeningWorkerState{State: "RUNNING", StartedAt: &started, Message: "正在获取日线并计算信号"} s.mu.Unlock() defer func() { finished := time.Now().UTC() s.mu.Lock() defer s.mu.Unlock() s.state.FinishedAt = &finished if processErr != nil { s.state.State = "FAILED" s.state.Message = "筛选未完成,请检查行情源与后台日志" } else { s.state.State = "COMPLETED" s.state.Message = fmt.Sprintf("检查 %d 只,完成 %d 只,生成 %d 条待审核候选,跳过 %d 只,失败 %d 只", s.state.Targets, s.state.Completed, s.state.Generated, s.state.Skipped, s.state.Failed) } }() c, err := s.market.configuration(ctx) if err != nil { return err } targets, err := s.repo.Targets(ctx, c) if err != nil { return err } // Rotate across eligible targets so an unavailable feed cannot starve other strategies. if len(targets) == 0 { return nil } s.mu.Lock() start := s.cursor % len(targets) count := min(20, len(targets)) s.state.Targets = count s.cursor = (start + count) % len(targets) s.mu.Unlock() for offset := 0; offset < count; offset++ { t := targets[(start+offset)%len(targets)] if ctx.Err() != nil { return ctx.Err() } now := time.Now().UTC() task, cancel := context.WithTimeout(ctx, 30*time.Second) h, e := s.market.adapter.History(task, c, t.Instrument.Symbol, t.Instrument.Exchange, now.AddDate(-3, 0, 0).Format("2006-01-02"), now.Format("2006-01-02")) if e == nil { var run model.StrategyScreening var draft *model.CandidateDraftInput run, draft, e = EvaluateScreening(t, c, h, time.Now().UTC()) if e == nil { e = s.repo.Record(task, t, run, draft) if e == nil { s.mu.Lock() s.state.Completed++ if draft != nil { s.state.Generated++ } s.mu.Unlock() } } } cancel() if e != nil { s.mu.Lock() if errors.Is(e, apperror.ErrConflict) || errors.Is(e, apperror.ErrUnavailable) { s.state.Skipped++ } else { s.state.Failed++ } s.mu.Unlock() } if e != nil && !errors.Is(e, apperror.ErrConflict) && !errors.Is(e, apperror.ErrUnavailable) && ctx.Err() == nil { logger.Warn("strategy screening failed", zap.String("strategy_version_id", t.Version.ID), zap.String("symbol", t.Instrument.Symbol), zap.Error(e)) } } return nil } func (s *ScreeningService) Run(ctx context.Context, logger *zap.Logger) { if s == nil || s.repo == nil || s.market == nil || s.market.adapter == nil || !s.development { return } for { if err := s.Process(ctx, logger); err != nil && !errors.Is(err, apperror.ErrUnavailable) && !errors.Is(err, apperror.ErrNotFound) && ctx.Err() == nil { logger.Warn("strategy screening worker failed", zap.Error(err)) } select { case <-ctx.Done(): return case <-s.wake: case <-time.After(5 * time.Minute): } } }