import WebSocket from 'isomorphic-ws' ; import EventEmitter from 'eventemitter3' ; import isBlob from 'is-blob' ; import urls from './urls.js' ; import { Bar , Channel , DataSource , DefaultCredentials , Quote , Trade , TradeUpdate , Message , } from './entities.js' ; export declare interface Events { open : ( stream : AlpacaStream ) => void ; close : ( stream : AlpacaStream ) => void ; authenticated : ( stream : AlpacaStream ) => void ; success : ( message : Message ) => void ; error : ( message : WebSocket . ErrorEvent ) => void ; subscription : ( message : Message ) => void ; message : ( message : Object ) => void ; trade_updates : ( update : TradeUpdate ) => void ; trade : ( trade : Trade ) => void ; quote : ( quote : Quote ) => void ; bar : ( bar : Bar ) => void ; } export declare interface AlpacaStream { on < U extends keyof Events > ( event , listener ) : this ; once < U extends keyof Events > ( event , listener ) : this ; emit < U extends keyof Events > ( event , ... args ) : boolean ; } export class AlpacaStream extends EventEmitter < string | symbol | any > { private host : string ; private connection : WebSocket ; private authenticated : boolean ; constructor ( protected params ) { super ( ) ; if ( ! ( 'paper' in params . credentials ) && ! ( 'key' in params . credentials && params . credentials . key . startsWith ( 'A' ) ) ) { params . credentials [ 'paper' ] = true ; } switch ( params . type ) { case 'account' : this . host = params . credentials . paper ? urls . websocket . account . replace ( 'api.' , 'paper-api.' ) : urls . websocket . account ; break ; case 'market_data' : this . host = urls . websocket . market_data ( this . params . source ) ; break ; default : this . host = 'unknown' ; } this . connection = new WebSocket ( this . host ) ; this . connection . onopen = ( ) => { let message = { } ; switch ( this . params . type ) { case 'account' : message = { action : 'authenticate' , data : { key_id : params . credentials . key , secret_key : params . credentials . secret , } , } ; break ; case 'market_data' : message = { action : 'auth' , ... params . credentials } ; break ; } this . connection . send ( JSON . stringify ( message ) ) ; this . emit ( 'open' , this ) ; } ; this . connection . onclose = ( ) => this . emit ( 'close' , this ) ; this . connection . onmessage = async ( event ) => { let data = event . data ; if ( isBlob ( data ) ) { data = await event . data . text ( ) ; } else if ( data instanceof ArrayBuffer ) { data = String . fromCharCode ( ... new Uint8Array ( event . data ) ) ; } let parsed = JSON . parse ( data ) , messages = this . params . type == 'account' ? [ parsed ] : parsed ; messages . forEach ( ( message ) => { this . emit ( 'message' , message ) ; if ( 'T' in message && message . msg == 'authenticated' ) { this . authenticated = true ; this . emit ( 'authenticated' , this ) ; } else if ( 'stream' in message && message . stream == 'authorization' ) { if ( message . data . status == 'authorized' ) { this . authenticated = true ; this . emit ( 'authenticated' , this ) ; } } if ( 'stream' in message && message . stream == 'trade_updates' ) { this . emit ( 'trade_updates' , message . data ) ; } const x = { success : 'success' , subscription : 'subscription' , error : 'error' , t : 'trade' , q : 'quote' , b : 'bar' , } ; if ( 'T' in message ) { this . emit ( x [ message . T . split ( '.' ) [ 0 ] ] , message ) ; } } ) ; } ; this . connection . onerror = ( err ) => { this . emit ( 'error' , err ) ; } ; } param channel param symbols subscribe ( channel , symbols = [ ] ) { switch ( this . params . type ) { case 'account' : this . send ( JSON . stringify ( { action : 'listen' , data : { streams : [ channel ] } } ) ) ; break ; case 'market_data' : let message = { action : 'subscribe' } ; message [ channel ] = symbols ; this . send ( JSON . stringify ( message ) ) ; break ; } return this ; } param channel param symbols unsubscribe ( channel , symbols = [ ] ) { switch ( this . params . type ) { case 'account' : this . send ( JSON . stringify ( { action : 'unlisten' , data : { streams : [ channel ] } } ) ) ; break ; case 'market_data' : let message = { action : 'unsubscribe' } ; message [ channel ] = symbols ; this . send ( JSON . stringify ( message ) ) ; break ; } return this ; } private send ( message ) { if ( ! this . authenticated ) { throw new Error ( 'not authenticated' ) ; } if ( typeof message == 'object' ) { message = JSON . stringify ( message ) ; } this . connection . send ( message ) ; return this ; } }