transparency.go raw

   1  package market
   2  
   3  import (
   4  	"fmt"
   5  	"sync"
   6  	"time"
   7  
   8  	"git.mleku.dev/mleku/dendrite/pkg/axiom"
   9  )
  10  
  11  // PriceFeed aggregates order books across venues and publishes a unified
  12  // view. This is the organism's first infrastructure service: making opaque
  13  // price structure visible. Revenue comes from subscribers who pay for the
  14  // consolidated feed, not from exploiting the opacity.
  15  type PriceFeed struct {
  16  	mu          sync.RWMutex
  17  	snapshots   map[string]*ConsolidatedSnapshot // asset → latest snapshot
  18  	subscribers []chan *ConsolidatedSnapshot
  19  	subMu       sync.Mutex
  20  }
  21  
  22  // ConsolidatedSnapshot is a single-asset view across all known venues.
  23  // It exposes the best available prices and the depth of each venue,
  24  // making cross-venue structure visible to any subscriber.
  25  type ConsolidatedSnapshot struct {
  26  	Asset    string
  27  	BestBid  float64
  28  	BidVenue Venue
  29  	BestAsk  float64
  30  	AskVenue Venue
  31  	Spread   float64
  32  	Venues   []VenueDepth
  33  	Time     time.Time
  34  }
  35  
  36  // VenueDepth is the depth summary for one venue.
  37  type VenueDepth struct {
  38  	Venue    Venue
  39  	BidDepth float64 // total bid volume
  40  	AskDepth float64 // total ask volume
  41  	Spread   float64 // local spread
  42  }
  43  
  44  // NewPriceFeed creates an empty consolidated price feed.
  45  func NewPriceFeed() *PriceFeed {
  46  	return &PriceFeed{
  47  		snapshots: make(map[string]*ConsolidatedSnapshot),
  48  	}
  49  }
  50  
  51  // Consolidate builds a unified snapshot from a VenueGraph's current state.
  52  func (pf *PriceFeed) Consolidate(vg *VenueGraph) {
  53  	assets := vg.Assets()
  54  
  55  	pf.mu.Lock()
  56  	defer pf.mu.Unlock()
  57  
  58  	for _, asset := range assets {
  59  		venues := vg.VenuesForAsset(asset)
  60  		if len(venues) == 0 {
  61  			continue
  62  		}
  63  
  64  		snap := &ConsolidatedSnapshot{
  65  			Asset: asset,
  66  			Time:  time.Now(),
  67  		}
  68  
  69  		for _, v := range venues {
  70  			book := vg.Book(asset, v)
  71  			if book == nil {
  72  				continue
  73  			}
  74  
  75  			bid := book.BestBid()
  76  			ask := book.BestAsk()
  77  
  78  			// Track best bid across venues.
  79  			if bid > snap.BestBid {
  80  				snap.BestBid = bid
  81  				snap.BidVenue = v
  82  			}
  83  
  84  			// Track best ask across venues.
  85  			if snap.BestAsk == 0 || (ask > 0 && ask < snap.BestAsk) {
  86  				snap.BestAsk = ask
  87  				snap.AskVenue = v
  88  			}
  89  
  90  			// Venue depth.
  91  			bidDepth := topVolume(book.Bids, len(book.Bids))
  92  			askDepth := topVolume(book.Asks, len(book.Asks))
  93  			snap.Venues = append(snap.Venues, VenueDepth{
  94  				Venue:    v,
  95  				BidDepth: bidDepth,
  96  				AskDepth: askDepth,
  97  				Spread:   book.Spread(),
  98  			})
  99  		}
 100  
 101  		if snap.BestAsk > 0 && snap.BestBid > 0 {
 102  			snap.Spread = snap.BestAsk - snap.BestBid
 103  		}
 104  
 105  		pf.snapshots[asset] = snap
 106  
 107  		// Publish to subscribers.
 108  		pf.subMu.Lock()
 109  		for _, ch := range pf.subscribers {
 110  			select {
 111  			case ch <- snap:
 112  			default: // drop if subscriber is slow
 113  			}
 114  		}
 115  		pf.subMu.Unlock()
 116  	}
 117  }
 118  
 119  // Subscribe returns a channel that receives consolidated snapshots.
 120  func (pf *PriceFeed) Subscribe() chan *ConsolidatedSnapshot {
 121  	ch := make(chan *ConsolidatedSnapshot, 32)
 122  	pf.subMu.Lock()
 123  	pf.subscribers = append(pf.subscribers, ch)
 124  	pf.subMu.Unlock()
 125  	return ch
 126  }
 127  
 128  // Snapshot returns the latest consolidated view for an asset.
 129  func (pf *PriceFeed) Snapshot(asset string) *ConsolidatedSnapshot {
 130  	pf.mu.RLock()
 131  	defer pf.mu.RUnlock()
 132  	return pf.snapshots[asset]
 133  }
 134  
 135  // Assets returns the list of assets with consolidated data.
 136  func (pf *PriceFeed) Assets() []string {
 137  	pf.mu.RLock()
 138  	defer pf.mu.RUnlock()
 139  	assets := make([]string, 0, len(pf.snapshots))
 140  	for a := range pf.snapshots {
 141  		assets = append(assets, a)
 142  	}
 143  	return assets
 144  }
 145  
 146  // ConsolidatedSnapshotToElements decomposes a consolidated snapshot into
 147  // lattice elements. The consolidated view is structure that replaces
 148  // the opacity — information that previously existed only in the gap
 149  // between venues.
 150  func ConsolidatedSnapshotToElements(cs *ConsolidatedSnapshot) []axiom.Element {
 151  	elems := []axiom.Element{
 152  		element{"asset", cs.Asset},
 153  		element{"consolidated-bid", cs.BestBid},
 154  		element{"consolidated-ask", cs.BestAsk},
 155  		element{"consolidated-spread", cs.Spread},
 156  		element{"bid-venue", string(cs.BidVenue)},
 157  		element{"ask-venue", string(cs.AskVenue)},
 158  		element{"venue-count", len(cs.Venues)},
 159  		element{"timestamp", fmt.Sprintf("%d", cs.Time.Unix())},
 160  	}
 161  	for _, vd := range cs.Venues {
 162  		elems = append(elems,
 163  			element{"venue-depth-bid", vd.BidDepth},
 164  			element{"venue-depth-ask", vd.AskDepth},
 165  		)
 166  	}
 167  	return elems
 168  }
 169