venue_graph.go raw

   1  package market
   2  
   3  import (
   4  	"fmt"
   5  	"os"
   6  	"sync"
   7  	"time"
   8  )
   9  
  10  // VenueGraph tracks order books across multiple venues for the same asset.
  11  // It detects cross-venue dislocations — prices that exist on one venue
  12  // but not another, or spreads that shouldn't exist.
  13  type VenueGraph struct {
  14  	mu    sync.RWMutex
  15  	books map[string]map[Venue]*OrderBook // asset → venue → latest book
  16  }
  17  
  18  // NewVenueGraph creates an empty cross-venue graph.
  19  func NewVenueGraph() *VenueGraph {
  20  	return &VenueGraph{
  21  		books: make(map[string]map[Venue]*OrderBook),
  22  	}
  23  }
  24  
  25  // Update records the latest order book for an asset/venue pair.
  26  func (vg *VenueGraph) Update(ob *OrderBook) {
  27  	vg.mu.Lock()
  28  	defer vg.mu.Unlock()
  29  	if vg.books[ob.Asset] == nil {
  30  		vg.books[ob.Asset] = make(map[Venue]*OrderBook)
  31  	}
  32  	vg.books[ob.Asset][ob.Venue] = ob
  33  }
  34  
  35  // DetectDislocations scans all tracked assets for cross-venue spread
  36  // opportunities. Returns dislocations where the spread exceeds the
  37  // given cost threshold.
  38  func (vg *VenueGraph) DetectDislocations(costThreshold float64) []CrossSpread {
  39  	vg.mu.RLock()
  40  	defer vg.mu.RUnlock()
  41  
  42  	var dislocations []CrossSpread
  43  
  44  	for _, venues := range vg.books {
  45  		// Compare all pairs of venues for this asset.
  46  		venueList := make([]Venue, 0, len(venues))
  47  		for v := range venues {
  48  			venueList = append(venueList, v)
  49  		}
  50  
  51  		for i := 0; i < len(venueList); i++ {
  52  			for j := i + 1; j < len(venueList); j++ {
  53  				bookA := venues[venueList[i]]
  54  				bookB := venues[venueList[j]]
  55  				cs := CompareBooksForSpread(bookA, bookB)
  56  
  57  				spread, _, _ := cs.BestSpread()
  58  				if spread > costThreshold {
  59  					dislocations = append(dislocations, *cs)
  60  				}
  61  			}
  62  		}
  63  	}
  64  
  65  	return dislocations
  66  }
  67  
  68  // Assets returns the list of tracked assets.
  69  func (vg *VenueGraph) Assets() []string {
  70  	vg.mu.RLock()
  71  	defer vg.mu.RUnlock()
  72  	assets := make([]string, 0, len(vg.books))
  73  	for a := range vg.books {
  74  		assets = append(assets, a)
  75  	}
  76  	return assets
  77  }
  78  
  79  // VenuesForAsset returns the venues tracking a specific asset.
  80  func (vg *VenueGraph) VenuesForAsset(asset string) []Venue {
  81  	vg.mu.RLock()
  82  	defer vg.mu.RUnlock()
  83  	venues := make([]Venue, 0, len(vg.books[asset]))
  84  	for v := range vg.books[asset] {
  85  		venues = append(venues, v)
  86  	}
  87  	return venues
  88  }
  89  
  90  // Book returns the latest order book for an asset/venue pair.
  91  func (vg *VenueGraph) Book(asset string, venue Venue) *OrderBook {
  92  	vg.mu.RLock()
  93  	defer vg.mu.RUnlock()
  94  	if vg.books[asset] == nil {
  95  		return nil
  96  	}
  97  	return vg.books[asset][venue]
  98  }
  99  
 100  // DislocationLog records detected cross-venue dislocations over time.
 101  type DislocationLog struct {
 102  	mu      sync.Mutex
 103  	entries []DislocationEntry
 104  	path    string
 105  }
 106  
 107  // DislocationEntry is a single recorded dislocation event.
 108  type DislocationEntry struct {
 109  	Time      time.Time
 110  	Asset     string
 111  	VenueA    Venue
 112  	VenueB    Venue
 113  	Spread    float64
 114  	BuyVenue  Venue
 115  	SellVenue Venue
 116  }
 117  
 118  // NewDislocationLog creates a log that writes to the given file path.
 119  func NewDislocationLog(path string) *DislocationLog {
 120  	return &DislocationLog{path: path}
 121  }
 122  
 123  // Record adds a dislocation entry and appends it to the log file.
 124  func (dl *DislocationLog) Record(cs *CrossSpread) {
 125  	spread, buyV, sellV := cs.BestSpread()
 126  	entry := DislocationEntry{
 127  		Time:      time.Now(),
 128  		Asset:     cs.Asset,
 129  		VenueA:    cs.VenueA,
 130  		VenueB:    cs.VenueB,
 131  		Spread:    spread,
 132  		BuyVenue:  buyV,
 133  		SellVenue: sellV,
 134  	}
 135  
 136  	dl.mu.Lock()
 137  	dl.entries = append(dl.entries, entry)
 138  	dl.mu.Unlock()
 139  
 140  	// Append to file.
 141  	if dl.path != "" {
 142  		f, err := os.OpenFile(dl.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
 143  		if err == nil {
 144  			fmt.Fprintf(f, "%s %s buy@%s sell@%s spread=%.2f\n",
 145  				entry.Time.Format(time.RFC3339), entry.Asset,
 146  				entry.BuyVenue, entry.SellVenue, entry.Spread)
 147  			f.Close()
 148  		}
 149  	}
 150  }
 151  
 152  // Entries returns all recorded dislocations.
 153  func (dl *DislocationLog) Entries() []DislocationEntry {
 154  	dl.mu.Lock()
 155  	defer dl.mu.Unlock()
 156  	out := make([]DislocationEntry, len(dl.entries))
 157  	copy(out, dl.entries)
 158  	return out
 159  }
 160  
 161  // Count returns the number of recorded dislocations.
 162  func (dl *DislocationLog) Count() int {
 163  	dl.mu.Lock()
 164  	defer dl.mu.Unlock()
 165  	return len(dl.entries)
 166  }
 167