exchange-monitor-go/trend_filter.go

package main

import (
	"encoding/json"
	"fmt"
	"io"
	"log"
	"math"
	"net/http"
	"os"
	"sort"
	"strings"
	"sync"
	"time"

)

// Binance klines endpoint (unauthenticated, USDⓈ-M futures)
const binanceFapiBase = "https://fapi.binance.com"
const klineCachePath = "data/kline_cache.json"
const signalsCachePath = "data/trend_signals_cache.json"

// Default thresholds for quiet/active classification.
var (
	range1hThreshold = 2.0 // 1h max range  < 2% → quiet
)

// FilterState stores trend filter results for one coin.
type FilterState struct {
	Coin          string  `json:"coin"`
	Range24h      float64 `json:"range_24h"`       // avg % range of 24 hourly candles
	Range1h       float64 `json:"range_1h"`         // max % range of 12 five-minute candles
	EMA52         float64 `json:"ema_52"`            // EMA52 of 1h close prices
	CurrentPrice  float64 `json:"current_price"`     // latest Binance price from live feed
	PriceAboveEMA bool    `json:"price_above_ema"`   // current price > EMA52
	Quiet24h      bool    `json:"quiet_24h"`          // 24h range below threshold
	Quiet1h       bool    `json:"quiet_1h"`           // 1h range below threshold
	FreshAnomaly  bool    `json:"fresh_anomaly"`      // coin in alert/confirmed state
	PassesFilter  bool    `json:"passes_filter"`      // all conditions met
	LastUpdated   int64   `json:"last_updated"`       // unix millis
	// v2: adaptive scoring fields
	VolumeRatio float64 `json:"volume_ratio"` // recent 1h volume / 24h avg volume
	EMASlope    float64 `json:"ema_slope"`    // EMA52 slope over last 3 candles (%)
	VolBaseline float64 `json:"vol_baseline"` // per-coin 24h volatility baseline (median %)
	SignalScore float64 `json:"signal_score"` // composite signal score 0-100
	Change1h    float64 `json:"change_1h"`    // 1h price change % (from 5m klines)
	KlineClose  float64 `json:"-"`            // last 5m kline close (for real-time drift calc, not serialized)
	DriftPct    float64 `json:"drift_pct"`    // real-time drift % from last kline close
}

// klineData holds parsed fields from one Binance kline.
type klineData struct {
	High   float64
	Low    float64
	Close  float64
	Volume float64
}

// TrendSignal records a signal event when trade conditions are met.
type TrendSignal struct {
	Timestamp    int64   `json:"timestamp"`
	Coin         string  `json:"coin"`
	Type         string  `json:"type"`    // "enter" or "exit"
	Category     string  `json:"category"` // "full" (anomaly+score>=70) or "high" (score>=90)
	SignalScore  float64 `json:"signal_score"`
	Price        float64 `json:"price"`
	EMA52        float64 `json:"ema_52"`
	EMASlope     float64 `json:"ema_slope"`
	VolumeRatio  float64 `json:"volume_ratio"`
	Range24h     float64 `json:"range_24h"`
	VolBaseline  float64 `json:"vol_baseline"`
	PriceAboveEMA bool   `json:"price_above_ema"`
	State        string  `json:"state"` // trend detector state at time of signal
}

// TrendFilter fetches Binance klines, computes EMA52/ranges, and filters
// coins that show fresh anomaly signals from TrendDetector.
type TrendFilter struct {
	mu            sync.RWMutex
	states        map[string]*FilterState
	store         *PriceStore
	trendDetector *TrendDetector
	client        *http.Client
	refreshTicker *time.Ticker
	stopCh        chan struct{}
	// Signal recording
	signals       []TrendSignal
	signaledCoins map[string]bool // coins currently in "full" enter signal state
	highScoreCoins map[string]bool // coins currently in "high" enter signal state
	OnNewSignal   func(TrendSignal) // callback for SSE broadcast
}

// NewTrendFilter creates a TrendFilter. Call Start() to begin periodic refresh.
func NewTrendFilter(store *PriceStore, td *TrendDetector) *TrendFilter {
	p := os.Getenv("HTTPS_PROXY")
	if p == "" {
		p = os.Getenv("https_proxy")
	}
	log.Printf("[TrendFilter] HTTPS_PROXY=%s", p)

	return &TrendFilter{
		client: &http.Client{
			Timeout:   15 * time.Second,
			Transport: &http.Transport{Proxy: http.ProxyFromEnvironment},
		},
		states:         make(map[string]*FilterState),
		store:          store,
		trendDetector:  td,
		stopCh:         make(chan struct{}),
		signaledCoins:  make(map[string]bool),
		highScoreCoins: make(map[string]bool),
	}
}

// Start begins the background kline fetch loop. The first fetch runs immediately.
func (tf *TrendFilter) Start() {
	// Load cached data on startup so we have data immediately
	if cached := loadCache(); cached != nil {
		tf.mu.Lock()
		tf.states = cached
		tf.mu.Unlock()
	}
	// Load signal history
	tf.loadSignals()

	go func() {
		tf.fetchBatch()
		tf.refreshTicker = time.NewTicker(5 * time.Minute)
		for {
			select {
			case <-tf.refreshTicker.C:
				tf.fetchBatch()
			case <-tf.stopCh:
				return
			}
		}
	}()
}

// Stop stops the background refresh.
func (tf *TrendFilter) Stop() {
	close(tf.stopCh)
	if tf.refreshTicker != nil {
		tf.refreshTicker.Stop()
	}
}

// Tick updates anomaly status from TrendDetector and recalculates PassesFilter.
// Call this every ~1s from the SSE broadcast loop (no HTTP calls).
func (tf *TrendFilter) Tick() {
	tf.mu.Lock()
	defer tf.mu.Unlock()

	for coin, st := range tf.states {
		// Update fresh anomaly from trend detector
		if tf.trendDetector != nil {
			st.FreshAnomaly = tf.trendDetector.IsAnomalous(coin)
		}

		// Update current price from live feed
		if p, ok := tf.store.Get(coin, ExBinance); ok && p > 0 {
			// Compute real-time drift from last kline close
			if st.KlineClose > 0 && st.CurrentPrice > 0 {
				st.DriftPct = math.Round((p-st.KlineClose)/st.KlineClose*10000) / 10000
			}
			st.CurrentPrice = p
			if st.EMA52 > 0 {
				st.PriceAboveEMA = p > st.EMA52
			}
		}

		// Re-evaluate overall pass
		st.PassesFilter = st.Quiet24h && st.Quiet1h && st.FreshAnomaly && st.PriceAboveEMA
		// Recalculate signal score (FreshAnomaly and PriceAboveEMA may have changed)
		st.SignalScore = computeSignalScore(st)
	}
	// Check for signal triggers (must hold lock)
	tf.checkSignals()
}

// Snapshot returns filter states, sorted with passing coins first.
func (tf *TrendFilter) Snapshot(limit int) []FilterState {
	tf.mu.RLock()
	defer tf.mu.RUnlock()

	result := make([]FilterState, 0, len(tf.states))
	for _, st := range tf.states {
		result = append(result, *st)
	}

	sort.Slice(result, func(i, j int) bool {
		// Highest signal score first
		if result[i].SignalScore != result[j].SignalScore {
			return result[i].SignalScore > result[j].SignalScore
		}
		return result[i].Coin < result[j].Coin
	})

	if limit > 0 && limit < len(result) {
		result = result[:limit]
	}
	return result
}

// fetchBatch fetches klines for all tracked coins concurrently.
func (tf *TrendFilter) fetchBatch() {
	log.Printf("[TrendFilter] Starting kline fetch (%d coins)...", len(TrackedCoins))
	// Collect coins that have a Binance symbol
	type coinSymbol struct {
		name   string
		symbol string
	}
	var targets []coinSymbol
	for _, tc := range TrackedCoins {
		if tc.BN != "" {
			targets = append(targets, coinSymbol{name: tc.Name, symbol: tc.BN})
		}
	}
	if len(targets) == 0 {
		return
	}

	// Semaphore: max 20 concurrent goroutines
	sem := make(chan struct{}, 20)
	var mu sync.Mutex
	type coinResult struct {
		name    string
		klines1h []klineData
		klines5m []klineData
		err     error
	}
	results := make([]coinResult, len(targets))

	var wg sync.WaitGroup
	for i, t := range targets {
		wg.Add(1)
		sem <- struct{}{}
		go func(idx int, coin, symbol string) {
			defer wg.Done()
			defer func() { <-sem }()

			k1h, err1 := tf.fetchKlines(symbol, "1h", 500)
			if err1 != nil {
				mu.Lock()
				results[idx] = coinResult{name: coin, err: err1}
				mu.Unlock()
				return
			}
			k5m, err2 := tf.fetchKlines(symbol, "5m", 12)
			if err2 != nil {
				mu.Lock()
				results[idx] = coinResult{name: coin, err: err2}
				mu.Unlock()
				return
			}
			mu.Lock()
			results[idx] = coinResult{name: coin, klines1h: k1h, klines5m: k5m}
			mu.Unlock()
		}(i, t.name, t.symbol)
	}
	wg.Wait()

	// Process results
	now := time.Now().UnixMilli()
	newStates := make(map[string]*FilterState, len(results))
	var errCount int
	for _, r := range results {
		if r.err != nil || len(r.klines1h) == 0 {
			if r.err != nil && errCount < 3 {
				log.Printf("[TrendFilter] Error for %s: %v", r.name, r.err)
				errCount++
			}
			continue
		}
		fs := tf.computeFilterState(r.name, r.klines1h, r.klines5m, now)
		newStates[r.name] = fs
	}

	// Merge: overwrite computed states, preserve anomaly for coins that errored
	tf.mu.Lock()
	for coin, fs := range newStates {
		tf.states[coin] = fs
		if tf.trendDetector != nil {
			fs.FreshAnomaly = tf.trendDetector.IsAnomalous(coin)
		}
		if p, ok := tf.store.Get(coin, ExBinance); ok && p > 0 {
			fs.CurrentPrice = p
			if fs.EMA52 > 0 {
				fs.PriceAboveEMA = p > fs.EMA52
			}
		}
		fs.PassesFilter = fs.Quiet24h && fs.Quiet1h && fs.FreshAnomaly && fs.PriceAboveEMA
		// Recalculate signal score with live data
		fs.SignalScore = computeSignalScore(fs)
	}
	tf.mu.Unlock()

	if len(newStates) > 0 {
		tf.saveCache()
	}
	log.Printf("[TrendFilter] Fetch complete: %d/%d coins have data", len(newStates), len(results))
}

// fetchKlines calls Binance fapi klines endpoint and parses the response.
func (tf *TrendFilter) fetchKlines(symbol, interval string, limit int) ([]klineData, error) {
	url := fmt.Sprintf("%s/fapi/v1/klines?symbol=%s&interval=%s&limit=%d",
		binanceFapiBase, strings.ToUpper(symbol), interval, limit)

	resp, err := tf.client.Get(url)
	if err != nil {
		return nil, fmt.Errorf("fetch %s: %w", symbol, err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("fetch %s: HTTP %d: %s", symbol, resp.StatusCode, string(body))
	}

	// Binance returns [[time,open,high,low,close,volume,...], ...]
	var raw [][]interface{}
	if err := json.NewDecoder(resp.Body).Decode(&raw); err != nil {
		return nil, fmt.Errorf("decode %s: %w", symbol, err)
	}

	result := make([]klineData, 0, len(raw))
	for _, item := range raw {
		if len(item) < 6 {
			continue
		}
		h := parseFloat(item[2])
		l := parseFloat(item[3])
		c := parseFloat(item[4])
		v := parseFloat(item[5])
		if h > 0 && l > 0 && c > 0 {
			result = append(result, klineData{High: h, Low: l, Close: c, Volume: v})
		}
	}
	return result, nil
}

// parseFloat converts a JSON string field to float64.
func parseFloat(v interface{}) float64 {
	s, _ := v.(string)
	var f float64
	_, _ = fmt.Sscanf(s, "%f", &f)
	return f
}

// computeFilterState derives all filter fields from kline data.
// v2: adaptive volatility baseline, volume ratio, EMA slope, composite score.
func (tf *TrendFilter) computeFilterState(coin string, k1h, k5m []klineData, now int64) *FilterState {
	fs := &FilterState{
		Coin:        coin,
		LastUpdated: now,
	}

	n := len(k1h)

	// ── Compute per-candle ranges for volatility baseline ──
	candleRanges := make([]float64, 0, n)
	for _, k := range k1h {
		if k.Low > 0 {
			candleRanges = append(candleRanges, (k.High-k.Low)/k.Low*100)
		}
	}

	// ── Adaptive volatility baseline ──
	// Short-term: median range of last 24 candles
	// Long-term:  median range of ALL candles
	// Quiet24h = short-term < long-term * 1.5
	if len(candleRanges) >= 24 {
		shortTerm := candleRanges
		if len(shortTerm) > 24 {
			shortTerm = candleRanges[len(candleRanges)-24:]
		}
		shortMedian := median(shortTerm)
		longMedian := median(candleRanges)

		fs.VolBaseline = math.Round(longMedian*100) / 100
		fs.Range24h = math.Round(shortMedian*100) / 100
		fs.Quiet24h = shortMedian < longMedian*1.5
	}

	// ── Compute EMA52 from hourly close prices ──
	if n >= 52 {
		prices := make([]float64, n)
		for i, k := range k1h {
			prices[i] = k.Close
		}
		ema := computeEMA(prices, 52)
		if len(ema) > 0 {
			fs.EMA52 = math.Round(ema[len(ema)-1]*10000) / 10000
		}
		// EMA slope: (ema[-1] - ema[-4]) / ema[-4] * 100
		if len(ema) >= 4 && ema[len(ema)-4] > 0 {
			slope := (ema[len(ema)-1] - ema[len(ema)-4]) / ema[len(ema)-4] * 100
			fs.EMASlope = math.Round(slope*10000) / 10000
		}
	}

	// ── Compute 1h max range from 5m klines ──
	if len(k5m) > 0 {
		var maxRange float64
		for _, k := range k5m {
			if k.Low <= 0 {
				continue
			}
			r := (k.High - k.Low) / k.Low * 100
			if r > maxRange {
				maxRange = r
			}
		}
		fs.Range1h = math.Round(maxRange*100) / 100
		fs.Quiet1h = fs.Range1h < range1hThreshold
	}

	// ── Volume ratio: last 1h volume / 24h avg volume ──
	if n >= 24 {
		lastVol := k1h[n-1].Volume
		var sumVol float64
		for i := n - 24; i < n-1; i++ {
			sumVol += k1h[i].Volume
		}
		avgVol := sumVol / 23 // exclude current candle from avg
		if avgVol > 0 {
			fs.VolumeRatio = math.Round(lastVol/avgVol*100) / 100
		}
	}

	// ── 1h price change % from 5m klines ──
	if len(k5m) >= 2 {
		firstClose := k5m[0].Close
		lastClose := k5m[len(k5m)-1].Close
		if firstClose > 0 {
			fs.Change1h = math.Round((lastClose-firstClose)/firstClose*10000) / 10000
		}
	}
	// Store last kline close for real-time drift calculation
	if len(k5m) > 0 {
		fs.KlineClose = k5m[len(k5m)-1].Close
	}

	// ── Composite SignalScore (0-100) ──
	var score float64
	if fs.PriceAboveEMA {
		score += 25
	}
	if fs.Quiet24h {
		score += 25
	}
	if fs.Quiet1h {
		score += 20
	}
	if fs.FreshAnomaly {
		score += 15
	}
	if fs.VolumeRatio > 1.5 {
		score += 15
	}
	// Bonus points
	if fs.EMASlope > 0.1 {
		score += 10
	}
	if fs.VolumeRatio > 3.0 {
		score += 10
	}
	// 1h price direction
	if fs.Change1h > 0 {
		score += 15
	} else if fs.Change1h < -0.1 {
		score -= 20
	} else if fs.Change1h < 0 {
		score -= 10
	}
	if score < 0 {
		score = 0
	}
	if score > 100 {
		score = 100
	}
	fs.SignalScore = score

	return fs
}

// median returns the median value of a sorted copy of the slice.
func median(values []float64) float64 {
	if len(values) == 0 {
		return 0
	}
	sorted := make([]float64, len(values))
	copy(sorted, values)
	sort.Float64s(sorted)
	mid := len(sorted) / 2
	if len(sorted)%2 == 0 {
		return (sorted[mid-1] + sorted[mid]) / 2
	}
	return sorted[mid]
}

// computeSignalScore calculates the composite signal score (0-100) from a FilterState.
// Must be called after FreshAnomaly, PriceAboveEMA, and volume fields are set.
func computeSignalScore(fs *FilterState) float64 {
	var score float64
	if fs.PriceAboveEMA {
		score += 25
	}
	if fs.Quiet24h {
		score += 25
	}
	if fs.Quiet1h {
		score += 20
	}
	if fs.FreshAnomaly {
		score += 15
	}
	if fs.VolumeRatio > 1.5 {
		score += 15
	}
	// Bonus: strong EMA uptrend
	if fs.EMASlope > 0.1 {
		score += 10
	}
	// Bonus: very high volume
	if fs.VolumeRatio > 3.0 {
		score += 10
	}
	// 1h price direction: positive change adds, negative change subtracts
	if fs.Change1h > 0 {
		score += 15
	} else if fs.Change1h < -0.1 {
		score -= 20 // actively dropping — heavily penalize
	} else if fs.Change1h < 0 {
		score -= 10 // slightly dropping
	}
	// Real-time drift: if current price is falling below last kline close, penalize
	if fs.DriftPct < -0.1 {
		score -= 15
	} else if fs.DriftPct < 0 {
		score -= 5
	}
	if score < 0 {
		score = 0
	}
	if score > 100 {
		score = 100
	}
	return score
}

// ── Signal recording ──

const signalScoreThreshold = 70.0
const highScoreThreshold = 90.0

// checkSignals scans all coins for enter/exit signal conditions.
// Must be called with tf.mu held.
func (tf *TrendFilter) checkSignals() {
	now := time.Now().UnixMilli()
	for coin, st := range tf.states {
		if st.EMA52 <= 0 {
			continue // no K-line data yet
		}

		// ── Full signal: FreshAnomaly + score >= 70 ──
		fullSignaled := tf.signaledCoins[coin]
		if st.FreshAnomaly && st.SignalScore >= signalScoreThreshold {
			if !fullSignaled {
				tf.recordSignal(now, coin, st, "enter", "full")
			}
		} else if fullSignaled {
			tf.recordSignal(now, coin, st, "exit", "full")
		}

		// ── High score signal: score >= 90 (no anomaly required) ──
		highSignaled := tf.highScoreCoins[coin]
		if st.SignalScore >= highScoreThreshold {
			if !highSignaled {
				tf.recordSignal(now, coin, st, "enter", "high")
			}
		} else if highSignaled {
			tf.recordSignal(now, coin, st, "exit", "high")
		}
	}
}

// recordSignal creates, stores, and broadcasts a signal event.
func (tf *TrendFilter) recordSignal(now int64, coin string, st *FilterState, sigType, category string) {
	sig := TrendSignal{
		Timestamp:     now,
		Coin:          coin,
		Type:          sigType,
		Category:      category,
		SignalScore:   st.SignalScore,
		Price:         st.CurrentPrice,
		EMA52:         st.EMA52,
		EMASlope:      st.EMASlope,
		VolumeRatio:   st.VolumeRatio,
		Range24h:      st.Range24h,
		VolBaseline:   st.VolBaseline,
		PriceAboveEMA: st.PriceAboveEMA,
		State:         tf.trendDetectorState(coin),
	}
	tf.signals = append(tf.signals, sig)

	// Track signaled state per category
	if sigType == "enter" {
		switch category {
		case "full":
			tf.signaledCoins[coin] = true
		case "high":
			tf.highScoreCoins[coin] = true
		}
	} else {
		switch category {
		case "full":
			delete(tf.signaledCoins, coin)
		case "high":
			delete(tf.highScoreCoins, coin)
		}
	}

	tf.saveSignals()
	if tf.OnNewSignal != nil {
		tf.OnNewSignal(sig)
	}
	log.Printf("[TrendFilter] SIGNAL %s/%s: %s score=%.0f price=%.4f vol=%.2fx slope=%.3f%%",
		sigType, category, coin, st.SignalScore, st.CurrentPrice, st.VolumeRatio, st.EMASlope)
}

// trendDetectorState reads the current trend detector state for a coin.
func (tf *TrendFilter) trendDetectorState(coin string) string {
	if tf.trendDetector == nil {
		return ""
	}
	return tf.trendDetector.State(coin)
}

// GetSignals returns the most recent N signals.
func (tf *TrendFilter) GetSignals(limit int) []TrendSignal {
	tf.mu.RLock()
	defer tf.mu.RUnlock()
	n := len(tf.signals)
	if n == 0 {
		return nil
	}
	start := n - limit
	if start < 0 {
		start = 0
	}
	result := make([]TrendSignal, n-start)
	copy(result, tf.signals[start:])
	// Return in reverse order (newest first)
	for i, j := 0, len(result)-1; i < j; i, j = i+1, j-1 {
		result[i], result[j] = result[j], result[i]
	}
	return result
}

// saveSignals persists signals to disk.
func (tf *TrendFilter) saveSignals() {
	if len(tf.signals) == 0 {
		return
	}
	// Keep only last 2000 signals
	if len(tf.signals) > 2000 {
		tf.signals = tf.signals[len(tf.signals)-2000:]
	}
	data, err := json.Marshal(tf.signals)
	if err != nil {
		return
	}
	os.WriteFile(signalsCachePath, data, 0644)
}

// loadSignals reads signals from disk.
func (tf *TrendFilter) loadSignals() {
	raw, err := os.ReadFile(signalsCachePath)
	if err != nil {
		return
	}
	var sigs []TrendSignal
	if err := json.Unmarshal(raw, &sigs); err != nil {
		return
	}
	tf.signals = sigs
	for _, s := range sigs {
		if s.Type == "enter" {
			switch s.Category {
			case "full":
				tf.signaledCoins[s.Coin] = true
			case "high":
				tf.highScoreCoins[s.Coin] = true
			}
		} else {
			switch s.Category {
			case "full":
				delete(tf.signaledCoins, s.Coin)
			case "high":
				delete(tf.highScoreCoins, s.Coin)
			}
		}
	}
	log.Printf("[TrendFilter] Loaded %d signal records", len(tf.signals))
}

// saveCache writes current filter states to disk (transient fields reset on load).
func (tf *TrendFilter) saveCache() {
	tf.mu.RLock()
	defer tf.mu.RUnlock()

	data, err := json.Marshal(tf.states)
	if err != nil {
		return
	}
	os.WriteFile(klineCachePath, data, 0644)
}

// loadCache reads filter states from disk, returning only non-transient fields.
func loadCache() map[string]*FilterState {
	raw, err := os.ReadFile(klineCachePath)
	if err != nil {
		return nil
	}
	var states map[string]*FilterState
	if err := json.Unmarshal(raw, &states); err != nil {
		return nil
	}
	// Reset transient fields — they will be set by Tick()
	for _, st := range states {
		st.CurrentPrice = 0
		st.PriceAboveEMA = false
		st.FreshAnomaly = false
		st.PassesFilter = false
	}
	log.Printf("[TrendFilter] Loaded %d coins from cache", len(states))
	return states
}

// computeEMA calculates EMA over price data for the given period.
// Uses SMA of first `period` values as seed, then EMA formula.
func computeEMA(prices []float64, period int) []float64 {
	if len(prices) < period || period < 2 {
		return nil
	}

	result := make([]float64, len(prices))

	// SMA seed
	var sum float64
	for i := 0; i < period; i++ {
		sum += prices[i]
	}
	result[period-1] = sum / float64(period)

	// EMA multiplier
	multiplier := 2.0 / float64(period+1)

	for i := period; i < len(prices); i++ {
		result[i] = (prices[i]-result[i-1])*multiplier + result[i-1]
	}

	// Fill leading entries with SMA value
	for i := 0; i < period-1; i++ {
		result[i] = result[period-1]
	}

	return result
}