package backtest import ( "context" "encoding/json" "errors" "fmt" "sort" "time" "nofx/kernel" "nofx/logger" "nofx/market" "nofx/store" ) func (r *Runner) loop(ctx context.Context) { defer close(r.doneCh) for { select { case <-ctx.Done(): r.handleStop(fmt.Errorf("context canceled: %w", ctx.Err())) return case <-r.stopCh: r.handleStop(nil) return case <-r.pauseCh: r.handlePause() <-r.resumeCh r.resumeFromPause() default: } err := r.stepOnce() if errors.Is(err, errBacktestCompleted) { r.handleCompletion() return } if errors.Is(err, errLiquidated) { r.handleLiquidation() return } if err != nil { r.handleFailure(err) return } } } func (r *Runner) stepOnce() error { state := r.snapshotState() if state.BarIndex >= r.feed.DecisionBarCount() { return errBacktestCompleted } ts := r.feed.DecisionTimestamp(state.BarIndex) marketData, multiTF, err := r.feed.BuildMarketData(ts) if err != nil { return err } priceMap := make(map[string]float64, len(marketData)) for symbol, data := range marketData { priceMap[symbol] = data.CurrentPrice } callCount := state.DecisionCycle + 1 shouldDecide := r.shouldTriggerDecision(state.BarIndex) var ( record *store.DecisionRecord decisionActions []store.DecisionAction tradeEvents = make([]TradeEvent, 0) execLog []string hadError bool ) decisionAttempted := shouldDecide if shouldDecide { ctx, rec, err := r.buildDecisionContext(ts, marketData, multiTF, priceMap, callCount) if err != nil { // Defensive nil check to prevent panic if buildDecisionContext returns error with nil record if rec != nil { rec.Success = false rec.ErrorMessage = fmt.Sprintf("failed to build trading context: %v", err) _ = r.logDecision(rec) } return err } record = rec var ( fullDecision *kernel.FullDecision fromCache bool cacheKey string ) if r.aiCache != nil { if key, err := computeCacheKey(ctx, r.cfg.PromptVariant, ts); err == nil { cacheKey = key if cached, ok := r.aiCache.Get(cacheKey); ok { fullDecision = cached fromCache = true } else if r.cfg.ReplayOnly { decisionErr := fmt.Errorf("replay_only enabled but cache miss at %d", ts) record.Success = false record.ErrorMessage = fmt.Sprintf("cached decision not found for ts=%d", ts) _ = r.logDecision(record) return decisionErr } } else { logger.Infof("failed to compute ai cache key: %v", err) } } if !fromCache { fd, err := r.invokeAIWithRetry(ctx) if err != nil { decisionAttempted = true hadError = true record.Success = false record.ErrorMessage = fmt.Sprintf("AI decision failed: %v", err) execLog = append(execLog, fmt.Sprintf("⚠️ AI decision failed: %v", err)) r.setLastError(err) } else { fullDecision = fd if r.cfg.CacheAI && r.aiCache != nil && cacheKey != "" { if err := r.aiCache.Put(cacheKey, r.cfg.PromptVariant, ts, fullDecision); err != nil { logger.Infof("failed to persist ai cache for %s: %v", r.cfg.RunID, err) } } } } if fullDecision != nil { r.fillDecisionRecord(record, fullDecision) sorted := sortDecisionsByPriority(fullDecision.Decisions) prevLogs := execLog decisionActions = make([]store.DecisionAction, 0, len(sorted)) execLog = make([]string, 0, len(sorted)+len(prevLogs)) if len(prevLogs) > 0 { execLog = append(execLog, prevLogs...) } for _, dec := range sorted { actionRecord, trades, logEntry, execErr := r.executeDecision(dec, priceMap, ts, callCount) if execErr != nil { actionRecord.Success = false actionRecord.Error = execErr.Error() hadError = true execLog = append(execLog, fmt.Sprintf("❌ %s %s: %v", dec.Symbol, dec.Action, execErr)) } else { actionRecord.Success = true execLog = append(execLog, fmt.Sprintf("✓ %s %s", dec.Symbol, dec.Action)) } if len(trades) > 0 { tradeEvents = append(tradeEvents, trades...) } if logEntry != "" { execLog = append(execLog, logEntry) } decisionActions = append(decisionActions, actionRecord) } } } cycleForLog := state.DecisionCycle if decisionAttempted { cycleForLog = callCount } liquidationEvents, liquidationNote, err := r.checkLiquidation(ts, priceMap, cycleForLog) if err != nil { if record != nil { record.Success = false record.ErrorMessage = err.Error() _ = r.logDecision(record) } return err } if len(liquidationEvents) > 0 { hadError = true tradeEvents = append(tradeEvents, liquidationEvents...) if record != nil { execLog = append(execLog, fmt.Sprintf("⚠️ Forced liquidation: %s", liquidationNote)) } } if record != nil { record.Decisions = decisionActions record.ExecutionLog = execLog record.Success = !hadError && liquidationNote == "" if liquidationNote != "" { record.ErrorMessage = liquidationNote } } equity, unrealized, _ := r.account.TotalEquity(priceMap) marginUsed := r.totalMarginUsed() r.updateState(ts, equity, unrealized, marginUsed, priceMap, decisionAttempted) snapshot := r.snapshotState() drawdownPct := 0.0 if snapshot.MaxEquity > 0 { drawdownPct = ((snapshot.MaxEquity - snapshot.Equity) / snapshot.MaxEquity) * 100 } equityPoint := EquityPoint{ Timestamp: ts, Equity: snapshot.Equity, Available: snapshot.Cash, PnL: snapshot.Equity - r.account.InitialBalance(), PnLPct: ((snapshot.Equity - r.account.InitialBalance()) / r.account.InitialBalance()) * 100, DrawdownPct: drawdownPct, Cycle: snapshot.DecisionCycle, } if err := appendEquityPoint(r.cfg.RunID, equityPoint); err != nil { return err } for _, evt := range tradeEvents { if err := appendTradeEvent(r.cfg.RunID, evt); err != nil { return err } } if record != nil { if err := r.logDecision(record); err != nil { return err } } if err := saveProgress(r.cfg.RunID, &snapshot, &r.cfg); err != nil { return err } if err := r.maybeCheckpoint(); err != nil { return err } r.persistMetadata() r.persistMetrics(false) if !hadError && liquidationNote == "" { r.setLastError(nil) } if snapshot.Liquidated { return errLiquidated } return nil } func (r *Runner) buildDecisionContext(ts int64, marketData map[string]*market.Data, multiTF map[string]map[string]*market.Data, priceMap map[string]float64, callCount int) (*kernel.Context, *store.DecisionRecord, error) { equity, unrealized, _ := r.account.TotalEquity(priceMap) available := r.account.Cash() marginUsed := r.totalMarginUsed() marginPct := 0.0 if equity > 0 { marginPct = (marginUsed / equity) * 100 } accountInfo := kernel.AccountInfo{ TotalEquity: equity, AvailableBalance: available, TotalPnL: equity - r.account.InitialBalance(), TotalPnLPct: ((equity - r.account.InitialBalance()) / r.account.InitialBalance()) * 100, MarginUsed: marginUsed, MarginUsedPct: marginPct, PositionCount: len(r.account.Positions()), } positions := r.convertPositions(priceMap) // Get candidate coins from strategy engine (includes source info) candidateCoins, err := r.strategyEngine.GetCandidateCoins() if err != nil { // Fallback to simple list if strategy engine fails candidateCoins = make([]kernel.CandidateCoin, 0, len(r.cfg.Symbols)) for _, sym := range r.cfg.Symbols { candidateCoins = append(candidateCoins, kernel.CandidateCoin{Symbol: sym, Sources: []string{"backtest"}}) } } runtime := int((ts - int64(r.cfg.StartTS*1000)) / 60000) ctx := &kernel.Context{ CurrentTime: time.UnixMilli(ts).UTC().Format("2006-01-02 15:04:05 UTC"), RuntimeMinutes: runtime, CallCount: callCount, Account: accountInfo, Positions: positions, CandidateCoins: candidateCoins, PromptVariant: r.cfg.PromptVariant, MarketDataMap: marketData, MultiTFMarket: multiTF, BTCETHLeverage: r.cfg.Leverage.BTCETHLeverage, AltcoinLeverage: r.cfg.Leverage.AltcoinLeverage, Timeframes: r.cfg.Timeframes, } // Fetch quantitative data if enabled in strategy (uses current data as approximation) strategyConfig := r.strategyEngine.GetConfig() if strategyConfig.Indicators.EnableQuantData { // Collect symbols to query (candidate coins + position coins) symbolSet := make(map[string]bool) for _, sym := range r.cfg.Symbols { symbolSet[sym] = true } for _, pos := range positions { symbolSet[pos.Symbol] = true } symbols := make([]string, 0, len(symbolSet)) for sym := range symbolSet { symbols = append(symbols, sym) } ctx.QuantDataMap = r.strategyEngine.FetchQuantDataBatch(symbols) if len(ctx.QuantDataMap) > 0 { logger.Infof("📊 Backtest: fetched quant data for %d symbols", len(ctx.QuantDataMap)) } } // Fetch OI ranking data if enabled in strategy (uses current data as approximation) if strategyConfig.Indicators.EnableOIRanking { ctx.OIRankingData = r.strategyEngine.FetchOIRankingData() if ctx.OIRankingData != nil { logger.Infof("📊 Backtest: OI ranking data ready: %d top, %d low positions", len(ctx.OIRankingData.TopPositions), len(ctx.OIRankingData.LowPositions)) } } // Fetch NetFlow ranking data if enabled in strategy if strategyConfig.Indicators.EnableNetFlowRanking { ctx.NetFlowRankingData = r.strategyEngine.FetchNetFlowRankingData() if ctx.NetFlowRankingData != nil { logger.Infof("💰 Backtest: NetFlow ranking data ready: inst_in=%d, inst_out=%d", len(ctx.NetFlowRankingData.InstitutionFutureTop), len(ctx.NetFlowRankingData.InstitutionFutureLow)) } } // Fetch Price ranking data if enabled in strategy if strategyConfig.Indicators.EnablePriceRanking { ctx.PriceRankingData = r.strategyEngine.FetchPriceRankingData() if ctx.PriceRankingData != nil { logger.Infof("📈 Backtest: Price ranking data ready for %d durations", len(ctx.PriceRankingData.Durations)) } } record := &store.DecisionRecord{ AccountState: store.AccountSnapshot{ TotalBalance: accountInfo.TotalEquity, AvailableBalance: accountInfo.AvailableBalance, TotalUnrealizedProfit: unrealized, PositionCount: accountInfo.PositionCount, MarginUsedPct: accountInfo.MarginUsedPct, }, CandidateCoins: make([]string, 0, len(candidateCoins)), Positions: r.snapshotPositions(priceMap), } for _, coin := range candidateCoins { record.CandidateCoins = append(record.CandidateCoins, coin.Symbol) } record.Timestamp = time.UnixMilli(ts).UTC() return ctx, record, nil } func (r *Runner) fillDecisionRecord(record *store.DecisionRecord, full *kernel.FullDecision) { record.InputPrompt = full.UserPrompt record.CoTTrace = full.CoTTrace if len(full.Decisions) > 0 { if data, err := json.MarshalIndent(full.Decisions, "", " "); err == nil { record.DecisionJSON = string(data) } } } func (r *Runner) invokeAIWithRetry(ctx *kernel.Context) (*kernel.FullDecision, error) { var lastErr error for attempt := 0; attempt < aiDecisionMaxRetries; attempt++ { // Use GetFullDecisionWithStrategy with the pre-configured strategy engine // This ensures backtest uses the same unified prompt generation as live trading fd, err := kernel.GetFullDecisionWithStrategy( ctx, r.mcpClient, r.strategyEngine, r.cfg.PromptVariant, ) if err == nil { return fd, nil } lastErr = err delay := time.Duration(attempt+1) * 500 * time.Millisecond time.Sleep(delay) } return nil, lastErr } func (r *Runner) shouldTriggerDecision(barIndex int) bool { if r.cfg.DecisionCadenceNBars <= 1 { return true } if barIndex < 0 { return true } return barIndex%r.cfg.DecisionCadenceNBars == 0 } func (r *Runner) updateState(ts int64, equity, unrealized, marginUsed float64, priceMap map[string]float64, advancedDecision bool) { r.stateMu.Lock() defer r.stateMu.Unlock() if r.state.MaxEquity == 0 || equity > r.state.MaxEquity { r.state.MaxEquity = equity } if r.state.MinEquity == 0 || equity < r.state.MinEquity { r.state.MinEquity = equity } if r.state.MaxEquity > 0 { drawdown := ((r.state.MaxEquity - equity) / r.state.MaxEquity) * 100 if drawdown > r.state.MaxDrawdownPct { r.state.MaxDrawdownPct = drawdown } } positions := make(map[string]PositionSnapshot) for _, pos := range r.account.Positions() { key := fmt.Sprintf("%s:%s", pos.Symbol, pos.Side) positions[key] = PositionSnapshot{ Symbol: pos.Symbol, Side: pos.Side, Quantity: pos.Quantity, AvgPrice: pos.EntryPrice, Leverage: pos.Leverage, LiquidationPrice: pos.LiquidationPrice, MarginUsed: pos.Margin, OpenTime: pos.OpenTime, AccumulatedFee: pos.AccumulatedFee, } } r.state.BarTimestamp = ts r.state.BarIndex++ if advancedDecision { r.state.DecisionCycle++ } r.state.Cash = r.account.Cash() r.state.Equity = equity r.state.UnrealizedPnL = unrealized r.state.RealizedPnL = r.account.RealizedPnL() r.state.Positions = positions r.state.LastUpdate = time.Now().UTC() } func (r *Runner) handleStop(reason error) { r.forceCheckpoint() if reason != nil { r.setLastError(reason) } else { r.setLastError(nil) } r.statusMu.Lock() r.err = reason r.status = RunStateStopped r.statusMu.Unlock() r.persistMetadata() r.persistMetrics(true) r.releaseLock() } func (r *Runner) handlePause() { r.forceCheckpoint() r.setLastError(nil) r.statusMu.Lock() r.status = RunStatePaused r.statusMu.Unlock() r.persistMetadata() r.persistMetrics(true) } func (r *Runner) resumeFromPause() { r.setLastError(nil) r.statusMu.Lock() r.status = RunStateRunning r.statusMu.Unlock() r.persistMetadata() } func (r *Runner) handleCompletion() { r.setLastError(nil) r.statusMu.Lock() r.status = RunStateCompleted r.statusMu.Unlock() r.persistMetadata() r.persistMetrics(true) r.releaseLock() } func (r *Runner) handleFailure(err error) { r.forceCheckpoint() if err != nil { r.setLastError(err) } r.statusMu.Lock() r.err = err r.status = RunStateFailed r.statusMu.Unlock() r.persistMetadata() r.persistMetrics(true) r.releaseLock() } func (r *Runner) handleLiquidation() { r.forceCheckpoint() r.setLastError(errLiquidated) r.statusMu.Lock() r.err = errLiquidated r.status = RunStateLiquidated r.statusMu.Unlock() r.persistMetadata() r.persistMetrics(true) r.releaseLock() } func sortDecisionsByPriority(decisions []kernel.Decision) []kernel.Decision { if len(decisions) <= 1 { return decisions } priority := func(action string) int { switch action { case "close_long", "close_short": return 1 case "open_long", "open_short": return 2 case "hold", "wait": return 3 default: return 99 } } result := make([]kernel.Decision, len(decisions)) copy(result, decisions) sort.Slice(result, func(i, j int) bool { pi := priority(result[i].Action) pj := priority(result[j].Action) if pi != pj { return pi < pj } return i < j }) return result }