gostock/internal/logic/strategy/boot.go

package strategy

import (
	"fmt"
	"log"
	"sync"

	"git.apinb.com/quant/gostock/internal/impl"
	"git.apinb.com/quant/gostock/internal/models"
)

func Boot() {
	InitCacheByAll()
}

// 启动 AI 分析任务
func BootAiStart(key string, ymd int) error {
	var datas []models.StratModel
	err := impl.DBService.Where("strat_key=? and ymd=? and ai_score=0", key, ymd).Find(&datas).Error
	if err != nil {
		log.Printf("Failed to query data: %v", err)
		return fmt.Errorf("query failed: %w", err)
	}

	// 构造任务列表
	var tasks []func()
	for _, row := range datas {
		row := row // 避免闭包捕获循环变量
		tasks = append(tasks, func() {
			BootAiTask(row.ID, row.Code)
		})
	}

	// 使用并发控制执行任务
	runWithLimit(20, tasks)
	return nil
}

// 执行单个 AI 分析任务
func BootAiTask(id uint, code string) {
	result, err := AiAnalysis(code)
	if err != nil {
		log.Printf("ERROR BootAiTask - ID: %d, Code: %s, Error: %v", id, code, err)
		return
	}
	log.Printf("SUCCESS BootAiTask - ID: %d, Code: %s", id, code)
	impl.DBService.Model(&models.StratModel{}).Where("id=?", id).Updates(result)
}

// 通用并发控制函数
func runWithLimit(limit int, tasks []func()) {
	semaphore := make(chan struct{}, limit)
	var wg sync.WaitGroup

	for _, task := range tasks {
		wg.Add(1)
		semaphore <- struct{}{} // 获取信号量
		go func(t func()) {
			defer wg.Done()
			defer func() { <-semaphore }() // 释放信号量
			t()
		}(task)
	}

	wg.Wait()
}