import * as _ from 'highland' ; import { createServer } from 'http' ; import { parse } from 'url' ; import { createStreamMerged , createStreamSorted } from './util/streams' ; import { createConsumerCreator } from './util/consumers' ; import { DevAlphaOptions , StreamAction , Strategy , RootState , Middleware , Consumer } from './types' ; import { rootReducer } from './reducers' ; import { createGuard , createStrategy , createBrokerRealtime , createBrokerBacktest } from './middleware' ; import { INITIALIZED , FINISHED , DASHBOARD_EVENTS , DASHBOARD_FINISHED , SOCKET_PORT } from './constants' ; export * from './constants' ; export * from './types' ; param { any } settings param { Strategy } strategy returns { Stream } export function createTrader ( settings , strategy ) { const config = { project : null , backtesting : true , client : null , startCapital : 0 , ... settings , initialStates : { ... settings . initialStates } , feeds : { ... settings . feeds } , backtest : { timestamp : 0 , commission : 0 , ... settings . backtest } , guard : { shorting : false , margin : false , restricted : [ ] , ... settings . guard } , dashboard : { active : false , port : SOCKET_PORT , ... settings . dashboard } } ; if ( typeof strategy !== 'function' ) { throw new Error ( 'strategy must be a function' ) ; } if ( config . dashboard . active && config . project === null ) { throw new Error ( 'the dashboard will not recognize your algorithm unless you set config.project to the ID of your DevAlpha project' ) ; } let state ; let reducing = false ; const store = { dispatch : ( action ) => input . write ( action ) , getState : ( ) => state , setState : ( nextState ) => { if ( ! reducing ) { throw new Error ( 'Cannot set state outside of reducer.' ) ; } state = nextState ; } } ; const createConsumer = createConsumerCreator ( store ) ; const reducerMiddleware = store => next => ( action ) => { reducing = true ; store . setState ( rootReducer ( store . getState ( ) , action ) ) ; reducing = false ; next ( action ) ; } ; const finishedConsumer = ( err , item , push , next ) => { if ( err ) { push ( err ) ; next ( ) ; } else if ( item === _ . nil ) { push ( null , { type : FINISHED , payload : { timestamp : config . backtesting === false ? Date . now ( ) : finishedAt } } ) ; push ( null , _ . nil ) ; } else { if ( config . backtesting !== false ) { finishedAt = ( < StreamAction > item ) . payload . timestamp ; } push ( null , item ) ; next ( ) ; } } ; const guardMiddleware = createGuard ( config . guard ) ; const strategyMiddleware = createStrategy ( strategy ) ; let brokerMiddleware ; if ( config . backtesting !== false || ! config . client ) { brokerMiddleware = createBrokerBacktest ( config . backtest . commission ) ; } else { brokerMiddleware = createBrokerRealtime ( config . client ) ; } let input ; let startedAt ; let finishedAt ; if ( config . backtesting === false ) { startedAt = Date . now ( ) ; finishedAt = Date . now ( ) ; input = createStreamMerged ( config . feeds ) ; } else { startedAt = config . backtest . timestamp ; finishedAt = config . backtest . timestamp ; input = createStreamSorted ( config . feeds ) ; } input . write ( { type : INITIALIZED , payload : { timestamp : startedAt , initialStates : config . initialStates , startCapital : config . startCapital } } ) ; const isValidAction = ( item ) => item . payload && typeof item . payload . timestamp !== 'undefined' ; let output = input . doto ( ( ) => { if ( config . backtesting !== false ) { input . pause ( ) ; } } ) . through < StreamAction , StreamAction > ( _ . seq ( ( _ as any ) . consume ( finishedConsumer ) , ( _ as any ) . filter ( isValidAction ) , ( _ as any ) . consume ( createConsumer ( guardMiddleware ) ) , ( _ as any ) . consume ( createConsumer ( brokerMiddleware ) ) , ( _ as any ) . consume ( createConsumer ( reducerMiddleware ) ) , ( _ as any ) . consume ( createConsumer ( strategyMiddleware ) ) ) as ( x : StreamAction ) => StreamAction ) . doto ( ( ) => { if ( config . backtesting !== false && input . _outgoing . length === 0 ) { input . resume ( ) ; } } ) . map ( action => ( { state : store . getState ( ) , action } ) ) ; if ( config . dashboard . active ) { const socketStream = output . fork ( ) ; output = output . fork ( ) ; let id = 0 ; const createMessage = ( message ) => { let response = '' ; response += `id: ${ id ++ }\n` ; response += `event: ${ message . type }\n` ; response += `data: ${ JSON . stringify ( message . payload ) }\n` ; response += `\n` ; return response ; } ; const app = createServer ( ( req , res ) => { const headers = { 'Access-Control-Allow-Origin' : '*' , 'Access-Control-Allow-Methods' : 'OPTIONS, GET' , 'Access-Control-Max-Age' : 60 * 60 * 24 * 30 , 'Cache-Control' : 'no-cache' , 'Connection' : 'close' } ; if ( req . method === 'OPTIONS' ) { res . writeHead ( 204 , headers ) ; return res . end ( ) ; } const url = parse ( req . url as string ) ; if ( url . pathname === `/${ config . project }` ) { headers [ 'Content-Type' ] = 'application/json' ; res . writeHead ( 200 , headers ) ; res . write ( JSON . stringify ( { message : 'ok' } ) ) ; res . end ( ) ; } else if ( url . pathname === `/${ config . project }/backtest` ) { headers [ 'Content-Type' ] = 'text/event-stream' , headers [ 'Connection' ] = 'keep-alive' , res . writeHead ( 200 , headers ) ; startedAt = Date . now ( ) ; socketStream . batchWithTimeOrCount ( 500 , 1000 ) . each ( ( events ) => { res . write ( createMessage ( { type : DASHBOARD_EVENTS , payload : { events } } ) ) ; } ) . done ( ( ) => { finishedAt = Date . now ( ) ; res . write ( createMessage ( { type : DASHBOARD_FINISHED , payload : { startedAt , finishedAt } } ) ) ; res . end ( ) ; app . close ( ) ; } ) ; } else { res . writeHead ( 404 , headers ) ; res . end ( ) ; } } ) ; app . listen ( config . dashboard . port ) ; } return output ; } export const devalpha = ( settings , strategy ) => { console . error ( 'the devalpha function is deprecated, please use createTrader() instead' ) ; return createTrader ( settings , strategy ) ; } ;