main.go

package main

import (
	"bufio"
	"context"
	"crypto/x509"
	"encoding/pem"
	"fmt"
	"log"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/joho/godotenv"
	"github.com/pranav-gunnala/4366-breakeven-traders-only/book"
	"github.com/pranav-gunnala/4366-breakeven-traders-only/client"
	"github.com/pranav-gunnala/4366-breakeven-traders-only/data"
	"github.com/pranav-gunnala/4366-breakeven-traders-only/frame"
)

var collector *data.Collector = nil

func collectUserInput() []string {
	reader := bufio.NewReader(os.Stdin)
	fmt.Print("Enter event IDs (comma-separated): ")
	raw, _ := reader.ReadString('\n')
	raw = strings.TrimSpace(raw)
	if raw == "" {
		log.Fatal("event IDs required")
	}
	var eventIDs []string
	for _, e := range strings.Split(raw, ",") {
		e = strings.TrimSpace(e)
		if strings.Contains(e, "kalshi.com") {
			eventIDs = append(eventIDs, strings.ToUpper(strings.Split(e, "/")[6]))
			continue
		}
		if e != "" {
			eventIDs = append(eventIDs, e)
		}
	}
	return eventIDs
}

func main() {
	godotenv.Load()

	//* the key data file used for frames.
	outPath := "./data/frames.jsonl"

	apiKey := os.Getenv("KALSHI_API_KEY")
	if apiKey == "" {
		log.Fatal("KALSHI_API_KEY not set")
	}

	pemBytes, err := os.ReadFile("keys/private.pem")
	if err != nil {
		log.Fatalf("read private key: %v", err)
	}
	block, _ := pem.Decode(pemBytes)
	if block == nil {
		log.Fatal("failed to decode PEM")
	}
	privateKey, err := x509.ParsePKCS1PrivateKey(block.Bytes)
	if err != nil {
		log.Fatalf("parse private key: %v", err)
	}
	restClient := client.NewHTTPClient(apiKey, privateKey)

	collector = data.NewCollector(restClient)

	var eventIDs []client.MarketMeta

	//* resolve event IDs -> market metadata via REST, followed by building tickers
	var tickers []string
	metaByTicker := make(map[string]client.MarketMeta)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	go collector.Collect(ctx)

	deadline := time.Now().Add(3 * time.Minute)
	for time.Now().Before(deadline) {
		eventIDs = data.GetMonitoring()
		if len(eventIDs) > 0 {
			break
		}
		time.Sleep(50 * time.Millisecond)
	}
	if len(eventIDs) == 0 {
		log.Fatal("collector-mode: no event IDs from open markets (timed out waiting for first fetch)")
	}

	for _, meta := range eventIDs {
		if meta.Ticker == "" {
			continue
		}
		tickers = append(tickers, meta.Ticker)
		metaByTicker[meta.Ticker] = meta
	}

	log.Printf("collector-mode: loaded %d markets/tickers", len(tickers))

	// Setup output writer
	writer, err := frame.NewWriter(outPath)
	if err != nil {
		log.Fatalf("open writer: %v", err)
	}
	defer writer.Close()

	//* store state per ticker
	var mu sync.Mutex
	books := make(map[string]*book.Book)
	trackers := make(map[string]*frame.MomentumTracker)
	for _, t := range tickers {
		books[t] = book.New(t)
		trackers[t] = frame.NewMomentumTracker()
	}

	//* Wire WS
	kalshi := client.NewClient(apiKey, privateKey)

	kalshi.OnSnapshot = func(snap client.OrderbookSnapshot) {
		mu.Lock()
		b, ok := books[snap.MarketTicker]
		mu.Unlock()

		if !ok {
			return
		}

		//* b was threadsafely acquired, it has its own lock
		b.ApplySnapshot(snap)
		log.Printf("[snapshot] %s seeded", snap.MarketTicker)
	}

	kalshi.OnDelta = func(delta client.OrderbookDelta) {
		mu.Lock()
		b, bOk := books[delta.MarketTicker]
		tr, tOk := trackers[delta.MarketTicker]
		mu.Unlock()

		if !bOk || !tOk {
			return
		}

		bbo, ofi, changed := b.ApplyDelta(delta)
		if !changed {
			return
		}
		if bbo.YesBid == 0 || bbo.YesAsk == 0 || bbo.YesBid >= bbo.YesAsk {
			return
		}

		// Compute T from close time
		meta := metaByTicker[delta.MarketTicker]
		T := 1.0 / 365.0 // fallback: 1 day
		if !meta.CloseTime.IsZero() && time.Now().Before(meta.CloseTime) {
			T = time.Until(meta.CloseTime).Hours() / 8760.0
		}
		nMarkets := meta.NMarkets
		if nMarkets == 0 {
			nMarkets = 2
		}

		momentum := tr.Push(bbo.Mid)
		fr := frame.Build(delta.MarketTicker, bbo, ofi, momentum, T, nMarkets)

		if err := writer.Write(fr); err != nil {
			log.Printf("[writer] error: %v", err)
		}
	}

	// kalshi.OnFill = func(fill client.Fill) {
	// } <-- legacy idea. no fill logic needed

	kalshi.OnError = func(err error) {
		log.Printf("[ws error] %v", err)
	}

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-sigCh
		fmt.Printf("\nShutting down... wrote %d frames\n", writer.Written())
		cancel()
	}()

	log.Println("Connecting to Kalshi WS...")
	if err := kalshi.Connect(ctx); err != nil {
		log.Fatalf("connect: %v", err)
	}
	log.Printf("Connected. Subscribing to %d tickers...", len(tickers))

	if err := kalshi.Subscribe(tickers); err != nil {
		log.Fatalf("subscribe: %v", err)
	}

	log.Printf("Subscribed. Writing BBO-change frames to %s", outPath)

	go func() {
		t := time.NewTicker(10 * time.Second)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				log.Printf("[stats] frames written: %d", writer.Written())
			}
		}
	}()

	<-ctx.Done()
	log.Printf("Done. Total frames: %d", writer.Written())
}