narrow down llm queries

Signed-off-by: kaedwen <kaedwen@heinrich.blue>

config.example.yaml

@@ -83,6 +83,8 @@ llm_scorer:
   max_retries: 2
   ensemble_weight: 0.7        # 0.0-1.0 (1.0 = LLM only, 0.7 = 70% LLM + 30% keyword)
   max_parallel: 20            # Max parallel LLM requests (worker pool size)
+  top_positive_count: 20      # Send top N positive articles to LLM (0 = all)
+  top_negative_count: 20      # Send top N negative articles to LLM (0 = all)
 
 # Logging
 log_level: info               # debug, info, warn, error

pkg/app/app.go

@@ -60,7 +60,6 @@ func New(cfg *config.Config, logger *slog.Logger) (*Application, error) {
 	if cfg.LLMScorer.Enabled {
 		llmScorer = news.NewLLMScorer(
 			&cfg.LLMScorer,
-			news.NewAnalyzer(), // Fallback analyzer
 			logger,
 		)
 		logger.Info("LLM scorer enabled",

pkg/app/news/aggregator.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"sort"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -185,6 +187,10 @@ func (a *Aggregator) llmWorker(ctx context.Context, workerID int) {
 				return
 			}
 
+			a.logger.Debug("worker received article",
+				slog.Int("worker_id", workerID),
+				slog.String("url", article.URL))
+
 			// Process article with LLM scoring
 			if err := a.llmScorer.Analyze(ctx, article); err != nil {
 				a.logger.Warn("LLM scoring failed",
@@ -209,7 +215,11 @@ func (a *Aggregator) llmWorker(ctx context.Context, workerID int) {
 func (a *Aggregator) fetchAllSources(ctx context.Context) error {
 	a.logger.Debug("fetching from all news sources")
 
-	var totalFetched atomic.Int64
+	var (
+		totalFetched  atomic.Int64
+		allArticles   []*model.NewsArticle
+		articlesMutex sync.Mutex
+	)
 
 	// Use errgroup for parallel fetching
 	g, gCtx := errgroup.WithContext(ctx)
@@ -233,36 +243,15 @@ func (a *Aggregator) fetchAllSources(ctx context.Context) error {
 
 			totalFetched.Add(int64(len(articles)))
 
-			// Process articles: keyword analysis + queue for LLM if enabled
+			// Keyword analysis + collect articles
 			for _, article := range articles {
 				// Always use keyword analyzer first (fast, synchronous)
 				a.analyzer.Analyze(article)
 
-				if a.llmScorer != nil && a.llmScorer.Enabled() {
+				// Collect for later selective processing
-					// Send to LLM worker pool for async processing
+				articlesMutex.Lock()
-					select {
+				allArticles = append(allArticles, article)
-					case a.llmJobs <- article:
+				articlesMutex.Unlock()
-						// Article queued successfully for LLM processing + DB insert
-					case <-sourceCtx.Done():
-						// Context canceled, store with keyword score only
-						article.SentimentMethod = "keyword"
-						if err := a.newsRepo.Create(sourceCtx, article); err != nil {
-							a.logger.Error("failed to store article",
-								slog.String("url", article.URL),
-								slog.Any("error", err),
-							)
-						}
-					}
-				} else {
-					// No LLM scorer, store immediately with keyword score
-					article.SentimentMethod = "keyword"
-					if err := a.newsRepo.Create(sourceCtx, article); err != nil {
-						a.logger.Error("failed to store article",
-							slog.String("url", article.URL),
-							slog.Any("error", err),
-						)
-					}
-				}
 			}
 
 			return nil
@@ -276,14 +265,69 @@ func (a *Aggregator) fetchAllSources(ctx context.Context) error {
 
 	fetched := int(totalFetched.Load())
 
-	a.logger.Info("news fetch completed",
+	// Phase 2: Selective LLM processing or direct storage
-		slog.Int("fetched", fetched),
+	if a.llmScorer != nil && a.llmScorer.Enabled() {
-		slog.Int("queued_for_llm", fetched), // All articles are queued for processing
+		// Select top N positive and top N negative articles for LLM
-	)
+		selectedForLLM := a.selectTopArticles(
+			allArticles,
+			a.llmScorer.cfg.TopPositiveCount,
+			a.llmScorer.cfg.TopNegativeCount,
+		)
 
-	// Notify listeners (workers will update DB asynchronously)
+		// Create a map for fast lookup
+		selectedMap := make(map[*model.NewsArticle]bool)
+		for _, article := range selectedForLLM {
+			selectedMap[article] = true
+		}
+
+		// Queue selected articles for LLM, store others immediately
+		keywordOnlyCount := 0
+		for _, article := range allArticles {
+			if selectedMap[article] {
+				// Send to LLM worker pool
+				select {
+				case a.llmJobs <- article:
+					// Article queued successfully for LLM processing + DB insert
+				case <-ctx.Done():
+					// Context canceled, store with keyword score only
+					if err := a.newsRepo.Create(ctx, article); err != nil {
+						a.logger.Error("failed to store article",
+							slog.String("url", article.URL),
+							slog.Any("error", err))
+					}
+				}
+			} else {
+				// Store with keyword score only
+				if err := a.newsRepo.Create(ctx, article); err != nil {
+					a.logger.Error("failed to store article",
+						slog.String("url", article.URL),
+						slog.Any("error", err))
+				}
+				keywordOnlyCount++
+			}
+		}
+
+		a.logger.Info("news fetch completed",
+			slog.Int("total_fetched", fetched),
+			slog.Int("selected_for_llm", len(selectedForLLM)),
+			slog.Int("keyword_only", keywordOnlyCount))
+	} else {
+		// No LLM scorer, store all with keyword scores
+		for _, article := range allArticles {
+			if err := a.newsRepo.Create(ctx, article); err != nil {
+				a.logger.Error("failed to store article",
+					slog.String("url", article.URL),
+					slog.Any("error", err))
+			}
+		}
+
+		a.logger.Info("news fetch completed",
+			slog.Int("fetched", fetched),
+			slog.Int("stored_with_keyword", fetched))
+	}
+
+	// Notify listeners
 	if fetched > 0 && a.onNewsUpdated != nil {
-		// Note: This fires immediately after fetch, not after all DB inserts
 		a.onNewsUpdated()
 	}
 
@@ -304,6 +348,64 @@ func (a *Aggregator) fetchAllSources(ctx context.Context) error {
 	return nil
 }
 
+// selectTopArticles picks the top N positive and top N negative articles
+// based on their keyword sentiment scores
+func (a *Aggregator) selectTopArticles(
+	articles []*model.NewsArticle,
+	topPositive, topNegative int,
+) []*model.NewsArticle {
+	// If both are 0 or negative, send all articles (backward compatibility)
+	if topPositive <= 0 && topNegative <= 0 {
+		return articles
+	}
+
+	// Separate into positive and negative based on keyword scores
+	var positive, negative []*model.NewsArticle
+	for _, article := range articles {
+		if article.SentimentScore == nil {
+			continue // Skip articles without scores
+		}
+		score := *article.SentimentScore
+		if score > 0 {
+			positive = append(positive, article)
+		} else if score < 0 {
+			negative = append(negative, article)
+		}
+		// Skip neutral (score == 0)
+	}
+
+	// Sort positive descending (most positive first)
+	sort.Slice(positive, func(i, j int) bool {
+		return *positive[i].SentimentScore > *positive[j].SentimentScore
+	})
+
+	// Sort negative ascending (most negative first)
+	sort.Slice(negative, func(i, j int) bool {
+		return *negative[i].SentimentScore < *negative[j].SentimentScore
+	})
+
+	// Select top N from each
+	selected := make([]*model.NewsArticle, 0)
+
+	if topPositive > 0 && len(positive) > 0 {
+		count := topPositive
+		if count > len(positive) {
+			count = len(positive)
+		}
+		selected = append(selected, positive[:count]...)
+	}
+
+	if topNegative > 0 && len(negative) > 0 {
+		count := topNegative
+		if count > len(negative) {
+			count = len(negative)
+		}
+		selected = append(selected, negative[:count]...)
+	}
+
+	return selected
+}
+
 func (a *Aggregator) GetRecent(ctx context.Context, limit int) ([]*db.NewsRepository, error) {
 	return nil, fmt.Errorf("not implemented")
 }

pkg/app/news/analyzer.go

@@ -10,6 +10,7 @@ import (
 type Analyzer struct {
 	positiveKeywords map[string]int
 	negativeKeywords map[string]int
+	isFallback       bool // If true, sets method to keyword_fallback instead of keyword
 }
 
 func NewAnalyzer() *Analyzer {
@@ -53,6 +54,13 @@ func NewAnalyzer() *Analyzer {
 	}
 }
 
+// NewFallbackAnalyzer creates an analyzer that marks results as keyword_fallback
+func NewFallbackAnalyzer() *Analyzer {
+	a := NewAnalyzer()
+	a.isFallback = true
+	return a
+}
+
 func (a *Analyzer) Analyze(article *model.NewsArticle) {
 	text := strings.ToLower(article.Title + " " + article.Content)
 	words := strings.Fields(text)
@@ -77,6 +85,7 @@ func (a *Analyzer) Analyze(article *model.NewsArticle) {
 		score := 0.0
 		article.SentimentScore = &score
 		article.SentimentLabel = "neutral"
+		article.SentimentMethod = model.SentimentMethodKeyword
 		return
 	}
 
@@ -91,4 +100,11 @@ func (a *Analyzer) Analyze(article *model.NewsArticle) {
 	} else {
 		article.SentimentLabel = "neutral"
 	}
+
+	// Set method based on whether this is a fallback analyzer
+	if a.isFallback {
+		article.SentimentMethod = model.SentimentMethodKeywordFallback
+	} else {
+		article.SentimentMethod = model.SentimentMethodKeyword
+	}
 }

pkg/app/news/llm_scorer.go

@@ -8,18 +8,26 @@ import (
 	"io"
 	"log/slog"
 	"net/http"
+	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/pheinrich/aitrade/pkg/config"
 	"github.com/pheinrich/aitrade/pkg/model"
 )
 
+// logDuration creates a human-readable duration attribute for logging
+func logDuration(key string, d time.Duration) slog.Attr {
+	return slog.String(key, d.Round(time.Millisecond).String())
+}
+
 // LLMScorer uses a local LLM (via Ollama) for contextual sentiment analysis
 type LLMScorer struct {
 	cfg              *config.LLMScorerConfig
 	httpClient       *http.Client
 	fallbackAnalyzer *Analyzer
 	logger           *slog.Logger
+	requestCounter   atomic.Uint64 // Unique ID for each request
 }
 
 // LLMRequest represents the Ollama API request
@@ -28,6 +36,7 @@ type LLMRequest struct {
 	Prompt      string  `json:"prompt"`
 	Temperature float64 `json:"temperature"`
 	Stream      bool    `json:"stream"`
+	Format      string  `json:"format,omitempty"` // "json" forces JSON-only output
 }
 
 // LLMResponse represents the Ollama API response
@@ -48,13 +57,12 @@ type SentimentResponse struct {
 
 func NewLLMScorer(
 	cfg *config.LLMScorerConfig,
-	fallbackAnalyzer *Analyzer,
 	logger *slog.Logger,
 ) *LLMScorer {
 	return &LLMScorer{
 		cfg:              cfg,
 		httpClient:       &http.Client{Timeout: cfg.Timeout.Duration},
-		fallbackAnalyzer: fallbackAnalyzer,
+		fallbackAnalyzer: NewFallbackAnalyzer(),
 		logger:           logger,
 	}
 }
@@ -79,9 +87,8 @@ func (l *LLMScorer) Analyze(ctx context.Context, article *model.NewsArticle) err
 			slog.String("article_id", fmt.Sprintf("%d", article.ID)),
 			slog.Any("error", err))
 
-		// Fallback to keyword analyzer
+		// Fallback to keyword analyzer (sets SentimentMethodKeywordFallback internally)
 		l.fallbackAnalyzer.Analyze(article)
-		article.SentimentMethod = "keyword_fallback"
 		return nil
 	}
 
@@ -90,7 +97,6 @@ func (l *LLMScorer) Analyze(ctx context.Context, article *model.NewsArticle) err
 		l.logger.Warn("LLM returned invalid score, using keyword fallback",
 			slog.Float64("score", sentimentResp.Score))
 		l.fallbackAnalyzer.Analyze(article)
-		article.SentimentMethod = "keyword_fallback"
 		return nil
 	}
 
@@ -109,7 +115,7 @@ func (l *LLMScorer) Analyze(ctx context.Context, article *model.NewsArticle) err
 				(keywordScore * (1.0 - l.cfg.EnsembleWeight))
 
 			article.SentimentScore = &finalScore
-			article.SentimentMethod = "ensemble"
+			article.SentimentMethod = model.SentimentMethodEnsemble
 
 			l.logger.Debug("ensemble scoring",
 				slog.Float64("llm_score", sentimentResp.Score),
@@ -118,12 +124,12 @@ func (l *LLMScorer) Analyze(ctx context.Context, article *model.NewsArticle) err
 				slog.Float64("weight", l.cfg.EnsembleWeight))
 		} else {
 			article.SentimentScore = &sentimentResp.Score
-			article.SentimentMethod = "llm"
+			article.SentimentMethod = model.SentimentMethodLLM
 		}
 	} else {
 		// LLM only
 		article.SentimentScore = &sentimentResp.Score
-		article.SentimentMethod = "llm"
+		article.SentimentMethod = model.SentimentMethodLLM
 	}
 
 	// Set sentiment label based on final score
@@ -169,11 +175,20 @@ JSON Response:`, article.Title, article.Content)
 
 // callOllamaAPI sends request to Ollama API and parses response
 func (l *LLMScorer) callOllamaAPI(ctx context.Context, prompt string) (sentiment *SentimentResponse, err error) {
+	// Generate unique request ID
+	reqID := l.requestCounter.Add(1)
+
+	// Create logger with request ID
+	reqLogger := l.logger.With(slog.Uint64("llm_req_id", reqID))
+
 	startTime := time.Now()
+	reqLogger.Debug("LLM request starting",
+		slog.String("model", l.cfg.ModelName))
+
 	defer func() {
 		duration := time.Since(startTime)
-		l.logger.Info("LLM request completed",
+		reqLogger.Info("LLM request completed",
-			slog.Duration("duration", duration),
+			logDuration("duration", duration),
 			slog.String("model", l.cfg.ModelName),
 			slog.Bool("success", err == nil))
 	}()
@@ -184,6 +199,7 @@ func (l *LLMScorer) callOllamaAPI(ctx context.Context, prompt string) (sentiment
 		Prompt:      prompt,
 		Temperature: l.cfg.Temperature,
 		Stream:      false,
+		Format:      "json", // Force JSON-only output (no markdown, no explanations)
 	}
 
 	jsonData, err := json.Marshal(reqBody)
@@ -222,6 +238,13 @@ func (l *LLMScorer) callOllamaAPI(ctx context.Context, prompt string) (sentiment
 		return nil, fmt.Errorf("failed to parse Ollama response: %w", err)
 	}
 
+	// Log raw LLM response for debugging (truncate if too long)
+	debugResp := llmResp.Response
+	if len(debugResp) > 500 {
+		debugResp = debugResp[:500] + "..."
+	}
+	l.logger.Debug("raw LLM response", slog.String("response", debugResp))
+
 	// Parse sentiment from LLM response
 	sentiment, err = l.parseSentiment(llmResp.Response)
 	return
@@ -229,27 +252,12 @@ func (l *LLMScorer) callOllamaAPI(ctx context.Context, prompt string) (sentiment
 
 // parseSentiment extracts structured sentiment from LLM text response
 func (l *LLMScorer) parseSentiment(response string) (*SentimentResponse, error) {
-	// Try to find JSON in response (LLM might add text before/after)
+	// With format: "json", Ollama returns pure JSON (no markdown, no extra text)
-	startIdx := -1
+	// But we still handle potential whitespace/newlines
-	endIdx := -1
+	response = strings.TrimSpace(response)
-
-	for i := 0; i < len(response); i++ {
-		if response[i] == '{' && startIdx == -1 {
-			startIdx = i
-		}
-		if response[i] == '}' {
-			endIdx = i + 1
-		}
-	}
-
-	if startIdx == -1 || endIdx == -1 {
-		return nil, fmt.Errorf("no JSON found in LLM response")
-	}
-
-	jsonStr := response[startIdx:endIdx]
 
 	var sentiment SentimentResponse
-	if err := json.Unmarshal([]byte(jsonStr), &sentiment); err != nil {
+	if err := json.Unmarshal([]byte(response), &sentiment); err != nil {
 		return nil, fmt.Errorf("failed to parse sentiment JSON: %w", err)
 	}
 

pkg/config/config.go

@@ -95,14 +95,16 @@ type NewsAuth struct {
 }
 
 type LLMScorerConfig struct {
-	Enabled        bool     `yaml:"enabled"`
+	Enabled          bool     `yaml:"enabled"`
-	Endpoint       string   `yaml:"endpoint"`
+	Endpoint         string   `yaml:"endpoint"`
-	ModelName      string   `yaml:"model_name"`
+	ModelName        string   `yaml:"model_name"`
-	Timeout        Duration `yaml:"timeout"`
+	Timeout          Duration `yaml:"timeout"`
-	Temperature    float64  `yaml:"temperature"`
+	Temperature      float64  `yaml:"temperature"`
-	MaxRetries     int      `yaml:"max_retries"`
+	MaxRetries       int      `yaml:"max_retries"`
-	EnsembleWeight float64  `yaml:"ensemble_weight"` // 0.0-1.0, 1.0 = LLM only
+	EnsembleWeight   float64  `yaml:"ensemble_weight"`   // 0.0-1.0, 1.0 = LLM only
-	MaxParallel    int      `yaml:"max_parallel"`    // Max parallel LLM requests (default: 20)
+	MaxParallel      int      `yaml:"max_parallel"`      // Max parallel LLM requests (default: 20)
+	TopPositiveCount int      `yaml:"top_positive_count"` // Send top N positive articles to LLM (0 = all)
+	TopNegativeCount int      `yaml:"top_negative_count"` // Send top N negative articles to LLM (0 = all)
 }
 
 func Load() (*Config, error) {
@@ -163,10 +165,12 @@ func Load() (*Config, error) {
 			Timeout: Duration{
 				Duration: time.Duration(getEnvInt("LLM_SCORER_TIMEOUT_SECONDS", 30)) * time.Second,
 			},
-			Temperature:    getEnvFloat("LLM_SCORER_TEMPERATURE", 0.3),
+			Temperature:      getEnvFloat("LLM_SCORER_TEMPERATURE", 0.3),
-			MaxRetries:     getEnvInt("LLM_SCORER_MAX_RETRIES", 2),
+			MaxRetries:       getEnvInt("LLM_SCORER_MAX_RETRIES", 2),
-			EnsembleWeight: getEnvFloat("LLM_SCORER_ENSEMBLE_WEIGHT", 0.7),
+			EnsembleWeight:   getEnvFloat("LLM_SCORER_ENSEMBLE_WEIGHT", 0.7),
-			MaxParallel:    getEnvInt("LLM_SCORER_MAX_PARALLEL", 20),
+			MaxParallel:      getEnvInt("LLM_SCORER_MAX_PARALLEL", 20),
+			TopPositiveCount: getEnvInt("LLM_SCORER_TOP_POSITIVE_COUNT", 20),
+			TopNegativeCount: getEnvInt("LLM_SCORER_TOP_NEGATIVE_COUNT", 20),
 		},
 		LogLevel: getEnv("LOG_LEVEL", "info"),
 	}

pkg/model/news.go

@@ -2,6 +2,16 @@ package model
 
 import "time"
 
+// SentimentMethod represents the method used for sentiment analysis
+type SentimentMethod string
+
+const (
+	SentimentMethodKeyword         SentimentMethod = "keyword"
+	SentimentMethodLLM             SentimentMethod = "llm"
+	SentimentMethodEnsemble        SentimentMethod = "ensemble"
+	SentimentMethodKeywordFallback SentimentMethod = "keyword_fallback"
+)
+
 type NewsArticle struct {
 	ID             int64
 	Source         string
@@ -17,5 +27,5 @@ type NewsArticle struct {
 	LLMSentimentScore *float64
 	LLMModel          *string
 	LLMConfidence     *float64
-	SentimentMethod   string // "keyword", "llm", "ensemble", "keyword_fallback"
+	SentimentMethod   SentimentMethod
 }