o geY@sZdZddlZddlZddlZddlZddlZddlZddlmZddl ZGdddZ dS)z} This file contains functions to stream data Coded by Tyler Bowers Github: https://github.com/tylerebowers/Schwab-API-Python N)sleepc @seZdZddZefddZedfdefddZed d dfdefd d Zd e fd dZ de e BfddZ de e BfddZ dAdefddZdBdedede fddZede eBfddZdCd!ee Bd"ee Bded#e fd$d%ZdCd!ee Bd"ee Bded#e fd&d'ZdCd!ee Bd"ee Bded#e fd(d)ZdCd!ee Bd"ee Bded#e fd*d+ZdCd!ee Bd"ee Bded#e fd,d-ZdCd!ee Bd"ee Bded#e fd.d/ZdCd!ee Bd"ee Bded#e fd0d1ZdCd!ee Bd"ee Bded#e fd2d3ZdCd!ee Bd"ee Bded#e fd4d5ZdCd!ee Bd"ee Bded#e fd6d7ZdCd!ee Bd"ee Bded#e fd8d9ZdCd!ee Bd"ee Bded#e fd:d;ZdDded#e fd?d@ZdS)EStreamcsDd_d_d_d_d_|_i_fdd}t|dS)z Initialize the stream object to stream data from Schwab Streamer :param client: Client object :type client: Client NrFcsjr dSdSN)activestopselfr//var/www/html/StreammingVPS/schwabdev/stream.py stop_atexit"s z$Stream.__init__..stop_atexit) _websocket_streamer_info _request_idr_thread_client subscriptionsatexitregister)r clientr rrr __init__s zStream.__init__c s|j}|jr|ddd|_ntddStjtj j } ztjtj j }|jj r5tdt j |jddd4IdH|_|jj rNtd |jd d |jjj|jd |jd dd}|jt|IdH||jIdHfi|d|_|jD]>\}}g}|D]\} } ||j|d| t| ddq|r|jtd|iIdH||jIdHfi|q ||jIdHfi|q1IdHswYWnttyV} zgd|_| t jjust| dkr |jj rtdWYd} ~ dS| t jjust| dkr'td| WYd} ~ dStjtj j |j dkrDtd| dWYd} ~ dStd| dWYd} ~ nd} ~ wwq$)z Start the streamer :param receiver_func: function to call when data is received :type receiver_func: function streamerInfoNrz&[Schwabdev] Could not get streamerInfoTz-[Schwabdev] Connecting to streaming server...streamerSocketUrl) ping_intervalz*[Schwabdev] Connected to streaming server.ADMINLOGINschwabClientChannelschwabClientFunctionId) AuthorizationSchwabClientChannelSchwabClientFunctionId)servicecommand parametersADDkeysfieldsrequestsFz'received 1000 (OK); then sent 1000 (OK)z%[Schwabdev] Stream connection closed.zno close frame received or sentz@[Schwabdev] Stream connection closed (likely no subscriptions): Zz2[Schwabdev] Stream has crashed within 90 seconds (zO), likely no subscriptions, invalid login, or lost connection (not restarting).z.[Schwabdev] Stream connection lost to server (z), reconnecting...)!r preferencesokjsongetr printdatetimenowtimezoneutcverbose websocketsconnectr basic_requesttokens access_tokensenddumpsrecvrritemsappendr_list_to_string Exception exceptionsConnectionClosedOKstrConnectionClosedErrorseconds) r receiver_funckwargsresponse start_time login_payloadr subsreqskeyr&errr _start_streamer(sn "    zStream._start_streamerTdaemonc sLjsfdd}tj||d_jdSjjr$tddSdS)z Start the stream :param receiver: function to call when data is received :type receiver: function :param daemon: whether to run the thread in the background (as a daemon) :type daemon: bool cstjfidSr)asynciorunrMrrEreceiverr rr _start_asyncosz"Stream.start.._start_asynctargetrNz"[Schwabdev] Stream already active.N)r threadingThreadrstartrr2r-)r rRrNrErSrrQr rXfs z Stream.startFc stjdddtjjdtjdddtjjd|r#tjdddtjjd|r/tjjjtjjdfdd }tj|d tj tjjjtjjdkr[ksbnt d d Sd S) ar Start the stream automatically at market open and close, will NOT erase subscriptions :param receiver: function to call when data is received :type receiver: function :param after_hours: include after hours trading :type after_hours: bool :param pre_hours: include pre hours trading :type pre_hours: bool rtzinfo ;cs tjtjj}|jtjjdkokno*d|ko(dkn}|rKjsKtj dkr?j j r?t dj d dn|s^jr^j j rXt djdd td q) NTr[rz6[Schwabdev] No subscriptions, starting stream anyways.)rRrNz[Schwabdev] Stopping Stream.F)clear_subscriptionsr)r.r/r0r1timereplaceweekdayrlenrrr2r-rXrr)r/in_hoursrNendrErRr rXrr checkersD   z"Stream.start_auto..checkerrTzU[Schwabdev] Stream was started outside of active hours and will launch when in hours.N) r.rcr0r1maxrdrVrWrXr/r-)r rR after_hours pre_hoursrNrErjrrhr start_autoxs 2 zStream.start_autorequestc Cs\dd}|dd}|dd}|dd}|dur||dg}||dg}||jvr4i|j|<|d krb|D]%}||j|vrK||j||<q:tt|t|j||B|j||<q:dS|d kryi|j|<|D] }||j||<qmdS|d kr|D]}||j|vr|j||qdS|d kr|j|D] }||j||<qdSdSdS) z Record the request into self.subscriptions (for the event of crashes) :param request: request :type request: dict cSs*t|tur |dSt|tur|SdS)N,)typerAsplitlist)strrr str_to_listsz+Stream._record_request..str_to_listr Nr!r"r%r&r#SUBSUNSUBSVIEW)r,rrssetpopr%) r rorur r!r"r%r&rKrrr _record_requests@     * zStream._record_requestr'cspfdd}t|tur|g}|D]}|qjr,td|i}t||dSjj r6t ddSdS)z Send a request to the stream :param requests: list of requests or a single request :type requests: list | dict csj|IdHdSr)r r8)to_sendrrr _sendszStream.send.._sendr'1[Schwabdev] Stream is not active, request queued.N) rqrsr{rr+r9rOrPrr2r-)r r'r}ror|rrr r8s   z Stream.sendcsjt|tur |g}|D]}||q |jr)td|i}|j|IdHdS|jj r3t ddSdS)z Send an async (must be awaited) request to the stream (functionally equivalent to send) :param requests: list of requests or a single request :type requests: list | dict r'Nr~) rqrsr{rr+r9r r8rr2r-)r r'ror|rrr send_asyncs  zStream.send_asyncracCs6|ri|_|jd7_||jdddd|_dS)z{ Stop the stream :param clear_subscriptions: clear records :type clear_subscriptions: bool rLOGOUT)r r!FN)rrr8r5r)r rarrr rs  z Stream.stopNr r!r"cCs|jdur|j}|jr|ddd|_ntdiS|dur3|D] }||dur2||=q'|jd7_| | |j|jd|jdd}|dur]t |dkr]||d <|S) a Create a basic request (all requests follow this format) :param service: service to use :type service: str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW"|"LOGIN"|"LOGOUT") :type command: str :param parameters: parameters to use :type parameters: dict :return: stream request :rtype: dict Nrrz*[Schwabdev] Could not use/get streamerInforschwabClientCustomerIdschwabClientCorrelId)r r! requestidSchwabClientCustomerIdSchwabClientCorrelIdr") r rr)r*r+r,r-r%rupperrf)r r r!r"rFrKrorrr r5s&    zStream.basic_requestlscCs0t|tur|St|turdtt|SdS)z Convert a list to a string (e.g. [1, "B", 3] -> "1,B,3"), or passthrough if already a string :param ls: list to convert :type ls: list | str :return: converted string :rtype: str rpN)rqrArsjoinmap)rrrr r= s zStream._list_to_stringr#r%r&returncC"|jd|t|t|ddS)ag Level one equities :param keys: list of keys to use (e.g. ["AMD", "INTC"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict LEVELONE_EQUITIESr$r"r5rr=r r%r&r!rrr level_one_equities," zStream.level_one_equitiescCr)a! Level one options, key format: [Underlying Symbol (6 characters including spaces) | Expiration (6 characters) | Call/Put (1 character) | Strike Price (5+3=8 characters)] :param keys: list of keys to use (e.g. ["GOOG 240809C00095000", "AAPL 240517P00190000"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict LEVELONE_OPTIONSr$rrrrrr level_one_options:rzStream.level_one_optionscCr)aR Level one futures, key format: '/' + 'root symbol' + 'month code' + 'year code'; month code is 1 character: (F: Jan, G: Feb, H: Mar, J: Apr, K: May, M: Jun, N: Jul, Q: Aug, U: Sep, V: Oct, X: Nov, Z: Dec), year code is 2 characters (i.e. 2024 = 24) :param keys: list of keys to use (e.g. ["/ESF24", "/GCG24"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict LEVELONE_FUTURESr$rrrrrr level_one_futuresHrzStream.level_one_futurescCr)a Level one futures options, key format: '.' + '/' + 'root symbol' + 'month code' + 'year code' + 'Call/Put code' + 'Strike Price' :param keys: list of keys to use (e.g. ["./OZCZ23C565"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict LEVELONE_FUTURES_OPTIONSr$rrrrrr level_one_futures_optionsVrz Stream.level_one_futures_optionscCr)a Level one forex, key format: 'from currency' + '/' + 'to currency' :param keys: list of keys to use (e.g. ["EUR/USD", "JPY/USD"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict LEVELONE_FOREXr$rrrrrr level_one_forexdrzStream.level_one_forexcCr)ab NYSE book orders :param keys: list of keys to use (e.g. ["NIO", "F"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict NYSE_BOOKr$rrrrrr nyse_bookrrzStream.nyse_bookcCr)ag NASDAQ book orders :param keys: list of keys to use (e.g. ["AMD", "CRWD"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict NASDAQ_BOOKr$rrrrrr nasdaq_bookrzStream.nasdaq_bookcCr)a Options book orders :param keys: list of keys to use (e.g. ["GOOG 240809C00095000", "AAPL 240517P00190000"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict OPTIONS_BOOKr$rrrrrr options_bookrzStream.options_bookcCr)ab Chart equity :param keys: list of keys to use (e.g. ["GOOG", "AAPL"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict CHART_EQUITYr$rrrrrr chart_equityrzStream.chart_equitycCr)aN Chart futures, key format: '/' + 'root symbol' + 'month code' + 'year code'; month code is 1 character: (F: Jan, G: Feb, H: Mar, J: Apr, K: May, M: Jun, N: Jul, Q: Aug, U: Sep, V: Oct, X: Nov, Z: Dec), year code is 2 characters (i.e. 2024 = 24) :param keys: list of keys to use (e.g. ["/ESF24", "/GCG24"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict CHART_FUTURESr$rrrrrr chart_futuresrzStream.chart_futurescCr)a Screener equity, key format: (PREFIX)_(SORTFIELD)_(FREQUENCY); Prefix: ($COMPX, $DJI, $SPX.X, INDEX_AL, NYSE, NASDAQ, OTCBB, EQUITY_ALL); Sortfield: (VOLUME, TRADES, PERCENT_CHANGE_UP, PERCENT_CHANGE_DOWN, AVERAGE_PERCENT_VOLUME), Frequency: (0 (all day), 1, 5, 10, 30 60) :param keys: list of keys to use (e.g. ["$DJI_PERCENT_CHANGE_UP_60", "NASDAQ_VOLUME_30"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict SCREENER_EQUITYr$rrrrrr screener_equityrzStream.screener_equitycCr)au Screener option key format: (PREFIX)_(SORTFIELD)_(FREQUENCY); Prefix: (OPTION_PUT, OPTION_CALL, OPTION_ALL); Sortfield: (VOLUME, TRADES, PERCENT_CHANGE_UP, PERCENT_CHANGE_DOWN, AVERAGE_PERCENT_VOLUME), Frequency: (0 (all day), 1, 5, 10, 30 60) :param keys: list of keys to use (e.g. ["OPTION_PUT_PERCENT_CHANGE_UP_60", "OPTION_CALL_TRADES_30"]) :type keys: list | str :param fields: list of fields to use :type fields: list | str :param command: command to use ("SUBS"|"ADD"|"UNSUBS"|"VIEW") :type command: str :return: stream request :rtype: dict SCREENER_OPTIONr$rrrrrr screener_optionsrzStream.screener_optionsAccount Activity0,1,2,3rvcCr)ap Account activity :param keys: list of keys to use (e.g. ["Account Activity"]) :type keys: list | str :param fields: list of fields to use (e.g. ["0,1,2,3"]) :type fields: list | str :param command: command to use ("SUBS"|"UNSUBS") :type command: str :return: stream request :rtype: dict ACCT_ACTIVITYr$rrrrrr account_activityrzStream.account_activity)Tr)r#)rrrv) __name__ __module__ __qualname__rr-rMboolrXrndictr{rsr8rrrAr5 staticmethodr=rrrrrrrrrrrrrrrrr rs2 >#& "$ $$$$$$$$$$$r) __doc__r+rrOr.rVr3rcrwebsockets.exceptionsrrrrr s