diff --git a/.gitignore b/.gitignore index c816185..ee318f7 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,3 @@ -.claude \ No newline at end of file +.claude +data/ +config.yaml \ No newline at end of file diff --git a/cmd/ibtest/main.go b/cmd/ibtest/main.go new file mode 100644 index 0000000..ecea4eb --- /dev/null +++ b/cmd/ibtest/main.go @@ -0,0 +1,228 @@ +package main + +import ( + "flag" + "fmt" + "log" + "os" + "os/signal" + "strconv" + "syscall" + "time" + + "github.com/scmhub/ibapi" +) + +type Wrapper struct { + ibapi.Wrapper + connected bool + nextID int64 + errorMsg string + marketDataRecv bool +} + +func (w *Wrapper) ConnectAck() { + fmt.Println("✓ Connected to IB Gateway") + w.connected = true +} + +func (w *Wrapper) NextValidID(orderId int64) { + fmt.Printf("✓ Next valid order ID: %d\n", orderId) + w.nextID = orderId +} + +func (w *Wrapper) ManagedAccounts(accountsList []string) { + fmt.Printf("✓ Managed accounts: %v\n", accountsList) +} + +func (w *Wrapper) Error(id int64, errorCode int64, errorMsgType int64, errorString string, advancedOrderRejectJson string) { + // Info messages (2100-2199) or connection status messages + if errorCode >= 2100 && errorCode < 2200 { + fmt.Printf("ℹ Info [%d]: %s\n", errorCode, errorString) + } else if errorCode >= 1100 && errorCode < 1300 { + // Warning messages (1100-1299) - system/connection warnings + fmt.Printf("⚠ Warning [%d]: %s\n", errorCode, errorString) + } else if errorCode > 1783000000000 { + // Very high error codes are typically connection status messages + // Ignore or log as debug + } else { + fmt.Printf("✗ Error [%d]: %s\n", errorCode, errorString) + w.errorMsg = errorString + } +} + +func (w *Wrapper) CurrentTime(t int64) { + serverTime := time.Unix(t, 0) + fmt.Printf("✓ Server time: %s\n", serverTime.Format("2006-01-02 15:04:05 MST")) +} + +func (w *Wrapper) ConnectionClosed() { + fmt.Println("⚠ Connection closed by server") + w.connected = false +} + +func (w *Wrapper) TickPrice(reqID int64, tickType ibapi.TickType, price float64, attrib ibapi.TickAttrib) { + w.marketDataRecv = true + tickName := map[ibapi.TickType]string{ + 1: "BID", 2: "ASK", 4: "LAST", 6: "HIGH", 7: "LOW", 9: "CLOSE", + 14: "OPEN", 15: "LOW_13_WEEK", 16: "HIGH_13_WEEK", 17: "LOW_26_WEEK", 18: "HIGH_26_WEEK", + 19: "LOW_52_WEEK", 20: "HIGH_52_WEEK", 35: "LAST_TIMESTAMP", 37: "MARK_PRICE", + 66: "BID_EXCH", 67: "ASK_EXCH", 68: "LAST_EXCH", 72: "HIGH_52W", 73: "LOW_52W", + 75: "PREV_CLOSE", 76: "HALTED", + } + name := tickName[tickType] + if name == "" { + name = fmt.Sprintf("Type_%d", tickType) + } + + timestamp := time.Now().Format("15:04:05") + fmt.Printf("[%s] %-22s %.2f\n", timestamp, name, price) +} + +func (w *Wrapper) TickSize(reqID int64, tickType ibapi.TickType, size ibapi.Decimal) { + w.marketDataRecv = true + tickName := map[ibapi.TickType]string{ + 0: "BID_SIZE", 3: "ASK_SIZE", 5: "LAST_SIZE", 8: "VOLUME", + 21: "AVG_VOLUME", 27: "CALL_OPTION_VOLUME", 28: "PUT_OPTION_VOLUME", + 29: "CALL_OPEN_INTEREST", 30: "PUT_OPEN_INTEREST", 34: "AUCTION_VOLUME", + 69: "BID_SIZE_EXCH", 70: "ASK_SIZE_EXCH", 71: "LAST_SIZE_EXCH", 74: "VOLUME_52W", + } + name := tickName[tickType] + if name == "" { + name = fmt.Sprintf("Type_%d", tickType) + } + + timestamp := time.Now().Format("15:04:05") + fmt.Printf("[%s] %-22s %s\n", timestamp, name, size.String()) +} + +func (w *Wrapper) TickString(reqID int64, tickType ibapi.TickType, value string) { + w.marketDataRecv = true + tickName := map[ibapi.TickType]string{ + 32: "BID_EXCH", 33: "ASK_EXCH", 45: "LAST_TIMESTAMP", 48: "RT_VOLUME", + 84: "LAST_EXCH", 85: "LAST_REG_TIME", 88: "DELAYED_LAST_TIMESTAMP", + } + name := tickName[tickType] + if name == "" { + name = fmt.Sprintf("Type_%d", tickType) + } + + // Convert Unix timestamp to ISO format for timestamp fields + displayValue := value + if tickType == 45 || tickType == 88 { // LAST_TIMESTAMP or DELAYED_LAST_TIMESTAMP + if ts, err := strconv.ParseInt(value, 10, 64); err == nil { + displayValue = time.Unix(ts, 0).Format(time.RFC3339) + } + } + + timestamp := time.Now().Format("15:04:05") + fmt.Printf("[%s] %-22s %s\n", timestamp, name, displayValue) +} + +func main() { + host := flag.String("host", "127.0.0.1", "IB Gateway host") + port := flag.Int("port", 4002, "IB Gateway port (4001 for live, 4002 for paper)") + clientID := flag.Int("client", 1, "Client ID") + timeout := flag.Int("timeout", 10, "Connection timeout in seconds") + symbol := flag.String("symbol", "", "Symbol to subscribe to market data (e.g. AAPL, TSLA)") + exchange := flag.String("exchange", "SMART", "Exchange for market data") + currency := flag.String("currency", "USD", "Currency for market data") + marketDataType := flag.Int("mdtype", 3, "Market data type: 1=Live, 2=Frozen, 3=Delayed(15min), 4=Delayed-Frozen") + flag.Parse() + + fmt.Printf("Testing IB Gateway connection...\n") + fmt.Printf("Host: %s:%d\n", *host, *port) + fmt.Printf("Client ID: %d\n\n", *clientID) + + wrapper := &Wrapper{} + client := ibapi.NewEClient(wrapper) + + // Connect + fmt.Println("→ Connecting...") + if err := client.Connect(*host, *port, int64(*clientID)); err != nil { + log.Fatalf("✗ Connection failed: %v\n", err) + } + defer client.Disconnect() + + // Wait for connection + start := time.Now() + for !wrapper.connected && time.Since(start) < time.Duration(*timeout)*time.Second { + time.Sleep(100 * time.Millisecond) + } + + if !wrapper.connected { + log.Fatal("✗ Connection timeout") + } + + // Wait for NextValidId + start = time.Now() + for wrapper.nextID == 0 && time.Since(start) < time.Duration(*timeout)*time.Second { + time.Sleep(100 * time.Millisecond) + } + + if wrapper.nextID == 0 { + log.Fatal("✗ Did not receive NextValidId") + } + + // Request server time + fmt.Println("\n→ Requesting server time...") + client.ReqCurrentTime() + + // Wait for response + time.Sleep(1 * time.Second) + + // Subscribe to market data if symbol provided + if *symbol != "" { + fmt.Printf("\n→ Subscribing to market data for %s...\n", *symbol) + + contract := &ibapi.Contract{ + Symbol: *symbol, + SecType: "STK", + Exchange: *exchange, + Currency: *currency, + } + + // Request market data type + // MarketDataType: 1=Live, 2=Frozen, 3=Delayed(15min), 4=Delayed-Frozen + mdTypeDesc := map[int]string{ + 1: "Live", + 2: "Frozen", + 3: "Delayed (15min)", + 4: "Delayed-Frozen", + } + fmt.Printf(" Market data type: %s\n", mdTypeDesc[*marketDataType]) + client.ReqMarketDataType(int64(*marketDataType)) + time.Sleep(500 * time.Millisecond) + + // Request market data + // ReqID: 1, Contract, GenericTickList: "", Snapshot: false, RegulatorySnapshot: false, MktDataOptions: nil + client.ReqMktData(1, contract, "", false, false, nil) + + // Wait for initial market data + fmt.Println(" Receiving market data... (Press Ctrl+C to exit)") + time.Sleep(2 * time.Second) + + if !wrapper.marketDataRecv { + fmt.Println(" ⚠ No market data received yet - may require market data subscription or market is closed") + } + + // Setup signal handler for graceful shutdown + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM) + + // Wait for Ctrl+C + <-sigChan + fmt.Println("\n\n→ Shutting down...") + + // Cancel market data subscription + client.CancelMktData(1) + fmt.Println(" Market data subscription cancelled") + } else { + fmt.Println("\n✓ Connection test successful!") + fmt.Println("\nConnection details:") + fmt.Printf(" - Next Order ID: %d\n", wrapper.nextID) + fmt.Printf(" - Client ID: %d\n", *clientID) + } + + os.Exit(0) +} diff --git a/config.yaml b/config.yaml deleted file mode 100644 index 31b3d00..0000000 --- a/config.yaml +++ /dev/null @@ -1,145 +0,0 @@ -# AI Trading Application Configuration -# This file can be placed at: -# - ./config.yaml (current directory) -# - ~/.config/aitrade/config.yaml -# - /etc/aitrade/config.yaml -# Or specify with: CONFIG_FILE=/path/to/config.yaml ./aitrade - -# Interactive Brokers Gateway -ib_gateway: - host: 127.0.0.1 - port: 4002 # 4002 for paper trading, 4001 for live - client_id: 777 # Use unique client ID (avoid 1, which may be used by other sessions) - market_data_type: 3 # 1=Live, 2=Frozen, 3=Delayed(15min), 4=Delayed-Frozen - -# Trading Configuration -trading: - # Strategy: defensive, normal, aggressive - strategy: normal - - # Stop Loss - stop_loss_enabled: true - stop_loss_percent: 3.0 - - # Rate Limiting - max_trades_per_hour: 6 - max_parallel_trades: 5 - pending_time: 300s # Seconds or Go duration (300, "5m", "1h30m") - - # Dry Run Mode (Paper Trading) - dry_run: false # Using IB Paper Trading account instead - dry_run_balance: 100000.0 - - # Auto Trading - trading_enabled: true - trading_interval: 60 # Seconds or Go duration (60, "1m", "5m") - watch_symbols: - - AAPL - - MSFT - - GOOGL - - TSLA - - AMZN - - # SELL Triggers - take_profit_percent: 5.0 - hold_time_minutes: 30 - sell_on_negative_sentiment: true - - # Trade Limits - max_trade_value: 2000.0 # 0 = unlimited - -# Database -database: - path: ./data/aitrade.db - -# Web Server -web: - host: 0.0.0.0 - port: "8080" - -# OpenID Connect (Optional) -oidc: - enabled: false - issuer: https://auth.example.com - client_id: aitrade - client_secret: your-secret-here - redirect_url: http://localhost:8080/oauth2/callback - scopes: - - openid - - profile - - email - -# News Aggregation -news: - poll_interval: 300s # Seconds or Go duration (300s = 5 minutes) - - # Default rate limits for all sources (can be overridden per source) - default_rate_limit: - max_per_hour: 100 # Poll every 5min = 12/hour, 100 gives plenty of buffer - max_per_day: 1000 - - # News Feed Sources - sources: - - name: "CNBC Top News" - url: "https://www.cnbc.com/id/100003114/device/rss/rss.html" - type: rss - enabled: true - - - name: "MarketWatch" - url: "https://feeds.marketwatch.com/marketwatch/realtimeheadlines" - type: rss - enabled: true - - - name: "Alpha Vantage News" - url: "https://www.alphavantage.co/query?function=NEWS_SENTIMENT&apikey=R8JA9C1DIEKM4SZM" - type: alphavantage - enabled: true - rate_limit: - max_per_hour: 5 # API key limit: 500/day - max_per_day: 500 - - - name: "Finnhub News" - url: "https://finnhub.io/api/v1/news?category=general&token=ca9mjbiad3ibg816q5b0" - type: finnhub - enabled: true - rate_limit: - max_per_hour: 60 # API key limit: 60/minute - max_per_day: 0 - - # Example: Feed with Basic Auth and Rate Limiting - # - name: "Reuters Business" - # url: "https://www.reuters.com/business/finance/rss" - # type: rss - # enabled: false - # auth: - # type: basic - # username: "your-username" - # password: "your-password" - # rate_limit: - # max_per_hour: 10 - # max_per_day: 100 - - # Example: Feed with custom headers (e.g., API key) - # - name: "Custom Feed" - # url: "https://example.com/feed" - # type: rss - # enabled: true - # headers: - # Authorization: "Bearer your-token" - # X-API-Key: "your-api-key" - # rate_limit: - # max_per_hour: 0 # Unlimited per hour - # max_per_day: 1000 # But limited to 1000/day - -# LLM Sentiment Scorer (Optional - requires Ollama) -llm_scorer: - enabled: true # Set to true to enable LLM-based sentiment - endpoint: http://192.168.40.133:11434 - model_name: mistral # "mistral", "llama2", "llama2:70b" - timeout: 120s # Seconds or Go duration (increased for CPU-only inference) - temperature: 0.3 # 0.0-1.0, lower = more deterministic - max_retries: 2 - ensemble_weight: 0.7 # 0.0-1.0 (1.0 = LLM only, 0.7 = 70% LLM + 30% keyword) - -# Logging -log_level: debug # debug, info, warn, error diff --git a/data/aitrade.db b/data/aitrade.db deleted file mode 100644 index c2da189..0000000 Binary files a/data/aitrade.db and /dev/null differ diff --git a/data/aitrade.db.backup.1782978610 b/data/aitrade.db.backup.1782978610 deleted file mode 100644 index 903461e..0000000 Binary files a/data/aitrade.db.backup.1782978610 and /dev/null differ diff --git a/pkg/app/app.go b/pkg/app/app.go index b510ad0..1258ccd 100644 --- a/pkg/app/app.go +++ b/pkg/app/app.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "time" "github.com/pheinrich/aitrade/pkg/app/client" "github.com/pheinrich/aitrade/pkg/app/news" @@ -60,7 +61,11 @@ func New(cfg *config.Config, logger *slog.Logger) (*Application, error) { } // Create news aggregator - newsAgg := news.NewAggregator(newsRepo, cfg.News.PollInterval.Duration, llmScorer, logger) + fetchTimeout := cfg.News.FetchTimeout.Duration + if fetchTimeout == 0 { + fetchTimeout = 5 * time.Minute // Default timeout + } + newsAgg := news.NewAggregator(newsRepo, cfg.News.PollInterval.Duration, fetchTimeout, llmScorer, logger) // Add news sources from config if len(cfg.News.Sources) > 0 { diff --git a/pkg/app/news/aggregator.go b/pkg/app/news/aggregator.go index f626f11..9a235d6 100644 --- a/pkg/app/news/aggregator.go +++ b/pkg/app/news/aggregator.go @@ -4,10 +4,12 @@ import ( "context" "fmt" "log/slog" + "sync/atomic" "time" "github.com/pheinrich/aitrade/pkg/config" "github.com/pheinrich/aitrade/pkg/db" + "golang.org/x/sync/errgroup" ) type Aggregator struct { @@ -16,6 +18,7 @@ type Aggregator struct { analyzer *Analyzer llmScorer *LLMScorer pollInterval time.Duration + fetchTimeout time.Duration logger *slog.Logger onNewsUpdated func() // Callback for SSE notifications } @@ -23,6 +26,7 @@ type Aggregator struct { func NewAggregator( newsRepo *db.NewsRepository, pollInterval time.Duration, + fetchTimeout time.Duration, llmScorer *LLMScorer, logger *slog.Logger, ) *Aggregator { @@ -32,6 +36,7 @@ func NewAggregator( analyzer: NewAnalyzer(), llmScorer: llmScorer, pollInterval: pollInterval, + fetchTimeout: fetchTimeout, logger: logger, } @@ -141,59 +146,79 @@ func (a *Aggregator) Run(ctx context.Context) error { func (a *Aggregator) fetchAllSources(ctx context.Context) error { a.logger.Debug("fetching from all news sources") - totalFetched := 0 - totalStored := 0 + var totalFetched, totalStored atomic.Int64 + // Use errgroup for parallel fetching + g, gCtx := errgroup.WithContext(ctx) + + // Launch goroutines for each source for _, source := range a.sources { - articles, err := source.Fetch(ctx) - if err != nil { - a.logger.Warn("failed to fetch from source", - slog.String("source", source.Name()), - slog.Any("error", err), - ) - continue - } + src := source // Capture loop variable + g.Go(func() error { + // Create a timeout context for this specific source + sourceCtx, cancel := context.WithTimeout(gCtx, a.fetchTimeout) + defer cancel() - totalFetched += len(articles) - - // Analyze sentiment and store - for _, article := range articles { - // Always use keyword analyzer first (fast) - a.analyzer.Analyze(article) - - // If LLM scorer enabled, use it (may fallback to keyword) - if a.llmScorer != nil && a.llmScorer.Enabled() { - if err := a.llmScorer.Analyze(ctx, article); err != nil { - a.logger.Warn("LLM scoring failed", - slog.String("url", article.URL), - slog.Any("error", err)) - // Article already has keyword sentiment, continue - } - } else { - // No LLM scorer, mark as keyword-only - article.SentimentMethod = "keyword" - } - - if err := a.newsRepo.Create(ctx, article); err != nil { - a.logger.Error("failed to store article", - slog.String("url", article.URL), + articles, err := src.Fetch(sourceCtx) + if err != nil { + a.logger.Warn("failed to fetch from source", + slog.String("source", src.Name()), slog.Any("error", err), ) - continue + return nil // Don't fail entire operation on single source error } - totalStored++ - } + totalFetched.Add(int64(len(articles))) + + // Analyze sentiment and store + for _, article := range articles { + // Always use keyword analyzer first (fast) + a.analyzer.Analyze(article) + + // If LLM scorer enabled, use it (may fallback to keyword) + if a.llmScorer != nil && a.llmScorer.Enabled() { + if err := a.llmScorer.Analyze(sourceCtx, article); err != nil { + a.logger.Warn("LLM scoring failed", + slog.String("url", article.URL), + slog.Any("error", err)) + // Article already has keyword sentiment, continue + } + } else { + // No LLM scorer, mark as keyword-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), + ) + continue + } + + totalStored.Add(1) + } + + return nil + }) } + // Wait for all goroutines to complete + if err := g.Wait(); err != nil { + return err + } + + fetched := int(totalFetched.Load()) + stored := int(totalStored.Load()) + a.logger.Info("news fetch completed", - slog.Int("fetched", totalFetched), - slog.Int("stored", totalStored), - slog.Int("duplicates", totalFetched-totalStored), + slog.Int("fetched", fetched), + slog.Int("stored", stored), + slog.Int("duplicates", fetched-stored), ) // Notify listeners if new articles were stored - if totalStored > 0 && a.onNewsUpdated != nil { + if stored > 0 && a.onNewsUpdated != nil { a.onNewsUpdated() } diff --git a/pkg/app/news/json_sources.go b/pkg/app/news/json_sources.go index 475a12a..c6b5b4a 100644 --- a/pkg/app/news/json_sources.go +++ b/pkg/app/news/json_sources.go @@ -40,10 +40,22 @@ type alphaVantageResponse struct { func NewAlphaVantageSource(name, url string, logger *slog.Logger) *AlphaVantageSource { now := time.Now() return &AlphaVantageSource{ - name: name, - url: url, - logger: logger, - httpClient: &http.Client{Timeout: 30 * time.Second}, + name: name, + url: url, + logger: logger, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= 10 { + return fmt.Errorf("stopped after 10 redirects") + } + logger.Debug("following redirect", + slog.String("source", name), + slog.String("from", via[len(via)-1].URL.String()), + slog.String("to", req.URL.String())) + return nil + }, + }, lastHourReset: now, lastDayReset: now, } @@ -181,10 +193,22 @@ type finnhubArticle struct { func NewFinnhubSource(name, url string, logger *slog.Logger) *FinnhubSource { now := time.Now() return &FinnhubSource{ - name: name, - url: url, - logger: logger, - httpClient: &http.Client{Timeout: 30 * time.Second}, + name: name, + url: url, + logger: logger, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= 10 { + return fmt.Errorf("stopped after 10 redirects") + } + logger.Debug("following redirect", + slog.String("source", name), + slog.String("from", via[len(via)-1].URL.String()), + slog.String("to", req.URL.String())) + return nil + }, + }, lastHourReset: now, lastDayReset: now, } diff --git a/pkg/app/news/sources.go b/pkg/app/news/sources.go index ccc1fc1..43b42a3 100644 --- a/pkg/app/news/sources.go +++ b/pkg/app/news/sources.go @@ -40,11 +40,24 @@ type RSSSource struct { func NewRSSSource(name, url string, logger *slog.Logger) *RSSSource { now := time.Now() return &RSSSource{ - name: name, - url: url, - parser: gofeed.NewParser(), - logger: logger, - httpClient: &http.Client{Timeout: 30 * time.Second}, + name: name, + url: url, + parser: gofeed.NewParser(), + logger: logger, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + // Follow up to 10 redirects (default behavior, made explicit) + if len(via) >= 10 { + return fmt.Errorf("stopped after 10 redirects") + } + logger.Debug("following redirect", + slog.String("source", name), + slog.String("from", via[len(via)-1].URL.String()), + slog.String("to", req.URL.String())) + return nil + }, + }, headers: make(map[string]string), lastHourReset: now, lastDayReset: now, diff --git a/pkg/config/config.go b/pkg/config/config.go index bd9c17b..57a1fb0 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -64,6 +64,7 @@ type OIDCConfig struct { type NewsConfig struct { PollInterval Duration `yaml:"poll_interval"` + FetchTimeout Duration `yaml:"fetch_timeout"` // Timeout per source (fetch + processing) DefaultRateLimit *RateLimit `yaml:"default_rate_limit,omitempty"` Sources []NewsSource `yaml:"sources"` }