"""Contains basic utilities to work with CQG OpenAPI COM component.""" import logging import threading from threading import Thread import pythoncom import win32com.client import win32event from win32com.client import constants win32com.client.gencache.EnsureModule("{51F35562-AEB4-4AB3-99E8-AC9666344B64}", 0, 4, 0) LOG_LEVEL = logging.DEBUG LOG_FORMAT = "%(asctime)s %(thread)d [%(levelname)s]: %(message)s" LOGGER_NAME = "CEL" def AssertMessage(condition, message): if not condition: raise RuntimeError(message) def GetLogger(): if not hasattr(GetLogger, "initialized"): logging.basicConfig(format=LOG_FORMAT) # set default format in case out logger is inaccessible # set-up CEL logger logger = logging.getLogger(LOGGER_NAME) handler = logging.StreamHandler() handler.setFormatter(logging.Formatter(LOG_FORMAT)) logger.addHandler(handler) logger.setLevel(LOG_LEVEL) logger.propagate = False GetLogger.initialized = True return logging.getLogger(LOGGER_NAME) # Class implements ICQGCELEvent interface. All sink classes used in examples should be inherited from this class. class CELSinkBase: def OnLineTimeChanged(self, newLineTime): pass def OnGWConnectionStatusChanged(self, newStatus): pass def OnDataConnectionStatusChanged(self, newStatus): pass def OnInstrumentSubscribed(self, symbol, cqgInstrument): pass def OnInstrumentChanged(self, cqgInstrument, cqgQuotes, cqgInstrumentProperties): pass def OnInstrumentDOMChanged(self, cqgInstrument, prevAsks, prevBids): pass def OnAccountChanged(self, changeType, cqgAccount, cqgPosition): pass def OnIsReady(self, readyStatus): pass def OnIdle(self): pass def OnDataError(self, cqgError, errorDescription): pass def OnCELStarted(self): pass def OnIncorrectSymbol(self, symbol): pass def OnCommodityInstrumentsResolved(self, commodityName, instrumentTypes, cqgCommodityInstruments): pass def OnGWEnvironmentChanged(self, eventCode, accountId, phase): pass def OnCurrencyRatesChanged(self, cqgCurrencyRates): pass def OnQueryProgress(self, cqgOrdersQuery, cqgError): pass def OnOrderChanged(self, changeType, cqgOrder, oldProperties, cqgFill, cqgError): pass def OnDataSourcesResolved(self, cqgDataSources, cqgError): pass def OnDataSourceSymbolsResolved(self, dataSourceAbbreviation, cqgDataSourceSymbols, cqgError): pass def OnCustomSessionsResolved(self, cqgSessionsCollection, cqgError): pass def OnTradableCommoditiesResolved(self, gwAccountId, cqgCommodities, cqgError): pass def OnTicksResolved(self, cqgTicks, cqgError): pass def OnTicksAdded(self, cqgTicks, addedTicksCount): pass def OnTicksRemoved(self, cqgTicks, removedTickIndex): pass def OnTimedBarsResolved(self, cqgTimedBars, cqgError): pass def OnTimedBarsAdded(self, cqgTimedBars): pass def OnTimedBarsUpdated(self, cqgTimedBars, index): pass def OnConstantVolumeBarsResolved(self, cqgConstantVolumeBars, cqgError): pass def OnConstantVolumeBarsAdded(self, cqgConstantVolumeBars): pass def OnConstantVolumeBarsUpdated(self, cqgConstantVolumeBars, index): pass def OnPointAndFigureBarsResolved(self, cqgPointAndFigureBars, cqgError): pass def OnPointAndFigureBarsAdded(self, cqgPointAndFigureBars): pass def OnPointAndFigureBarsUpdated(self, cqgPointAndFigureBars, index): pass def OnYieldsResolved(self, cqgYields, cqgError): pass def OnYieldsAdded(self, cqgYields): pass def OnYieldsUpdated(self, cqgYields, index): pass def OnTFlowBarsResolved(self, cqgTFlowBars, cqgError): pass def OnTFlowBarsAdded(self, cqgTFlowBars): pass def OnTFlowBarsUpdated(self, cqgTFlowBars, index): pass def OnCustomStudyDefinitionsResolved(self, cqgCustomStudyDefinitions, cqgError): pass def OnTradingSystemDefinitionsResolved(self, cqgCustomStudyDefinitions, cqgError): pass def OnConditionDefinitionsResolved(self, cqgConditionDefinitions, cqgError): pass def OnQFormulaDefinitionsResolved(self, cqgQFormulaDefinitions, cqgError): pass def OnCustomStudyResolved(self, cqgCustomStudy, cqgError): pass def OnCustomStudyAdded(self, cqgCustomStudy): pass def OnCustomStudyUpdated(self, cqgCustomStudy, index): pass def OnConditionResolved(self, cqgCondition, cqgError): pass def OnConditionAdded(self, cqgCondition): pass def OnConditionUpdated(self, cqgCondition, index): pass def OnTradingSystemResolved(self, cqgTradingSystem, cqgError): pass def OnTradingSystemAddNotification(self, cqgTradingSystem, cqgTradingSystemAddInfo): pass def OnTradingSystemUpdateNotification(self, cqgTradingSystem, cqgTradingSystemUpdateInfo): pass def OnExpressionResolved(self, cqgExpression, cqgError): pass def OnExpressionAdded(self, cqgExpression): pass def OnExpressionUpdated(self, cqgExpression, index): pass def OnAlgorithmicOrderRegistrationComplete(self, guid, cqgError): pass def OnAlgorithmicOrderPlaced(self, guid, mainParams, customProps): pass def OnTimedBarsInserted(self, cqgTimedBars, index): pass def OnTimedBarsRemoved(self, cqgTimedBars, index): pass def OnConstantVolumeBarsInserted(self, cqgConstantVolumeBars, index): pass def OnConstantVolumeBarsRemoved(self, cqgConstantVolumeBars, index): pass def OnPointAndFigureBarsInserted(self, cqgPointAndFigureBars, index): pass def OnPointAndFigureBarsRemoved(self, cqgPointAndFigureBars, index): pass def OnYieldsInserted(self, cqgYields, index): pass def OnYieldsRemoved(self, cqgYields, index): pass def OnTFlowBarsInserted(self, cqgTFlowBars, index): pass def OnTFlowBarsRemoved(self, cqgTFlowBars, index): pass def OnExpressionInserted(self, cqgExpression, index): pass def OnExpressionRemoved(self, cqgExpression, index): pass def OnConditionInserted(self, cqgCondition, index): pass def OnConditionRemoved(self, cqgCondition, index): pass def OnCustomStudyInserted(self, cqgCustomStudy, index): pass def OnCustomStudyRemoved(self, cqgCustomStudy, index): pass def OnICConnectionStatusChanged(self, newStatus): pass def OnAllOrdersCanceled(self, orderSide, gwAccountIds, instrumentNames): pass def OnInstrumentsGroupResolved(self, cqgInstrumentsGroup, cqgError): pass def OnInstrumentsGroupChanged(self, changeType, cqgInstrumentsGroup, instrumentsNames): pass def OnBarsTimestampsResolved(self, cqgBarsTimestamps, cqgError): pass def OnStrategyDefinitionProgress(self, cqgDefinition, cqgError): pass def OnTradingSystemInsertNotification(self, cqgTradingSystem, cqgTradingSystemInsertInfo): pass def OnTradingSystemRemoveNotification(self, cqgTradingSystem, cqgTradingSystemRemoveInfo): pass def OnTradingSystemTradeRelationAddNotification(self, cqgTradingSystem, cqgTradingSystemRelationAddInfo): pass def OnTradableExchangesResolved(self, gwAccountId, cqgExchanges, cqgError): pass def OnHistoricalSessionsResolved(self, cqgHistoricalSessions, cqgHistoricalSessionsRequest, cqgError): pass def OnSummariesStatementResolved(self, cqgSummariesStatement, cqgError): pass def OnPositionsStatementResolved(self, cqgPositionsStatement, cqgError): pass def OnManualFillUpdateResolved(self, cqgManualFillRequest, cqgError): pass def OnManualFillsResolved(self, cqgManualFills, cqgError): pass def OnManualFillChanged(self, cqgManualFill, modifyType): pass def OnAuthenticationStatusChanged(self, newStatus, cqgError): pass def OnPasswordChanged(self, requestStatus, cqgError): pass def OnAdvancedStudyDefinitionsResolved(self, cqgAdvancedStudyDefinitions, cqgError): pass def OnAdvancedStudyResolved(self, cqgAdvancedStudy, cqgError): pass def OnAdvancedStudyAdded(self, cqgAdvancedStudy): pass def OnAdvancedStudyUpdated(self, cqgAdvancedStudy, index): pass def OnAdvancedStudyInserted(self, cqgAdvancedStudy, index): pass def OnAdvancedStudyRemoved(self, cqgAdvancedStudy, index): pass def OnSubMinuteBarsResolved(self, cqgSubminuteBars, cqgError): pass def OnSubMinuteBarsAdded(self, cqgSubminuteBars): pass def OnSubMinuteBarsUpdated(self, cqgSubminuteBars, index): pass def OnSubMinuteBarsInserted(self, cqgSubminuteBars, index): pass def OnSubMinuteBarsRemoved(self, cqgSubminuteBars, index): pass def OnInstrumentResolved(self, symbol, cqgInstrument, cqgError): pass def OnStrategyQuoteRequestResolved(self, cqgRFQ, isProcessed, cqgError): pass def OnQFormulaDefinitionResolved(self, cqgQFormulaDefinition, cqgError): pass def OnQFormulaCreated(self, name, cqgQFormulaDefinition, cqgError): pass def OnQFormulaDeleted(self, name, cqgError): pass def OnQFormulaDefinitionResolvedEx(self, requestStr, cqgQFormulaDefinition, cqgError): pass def OnExecutionPatternReady(self, requestId, executionPatternString, cqgError): pass # Auxiliary class class SinkInternal(CELSinkBase): logFormat = "%(asctime)s %(thread)d [%(levelname)s] %(name)s: %(message)s" logLevel = LOG_LEVEL # define static variable def __init__(self): self.logLevel = LOG_LEVEL # set again in case if global has changed self.logger = logging.getLogger(SinkInternal.__name__) loggerHandler = logging.StreamHandler() loggerHandler.setFormatter(logging.Formatter(self.logFormat)) self.logger.addHandler(loggerHandler) self.logger.setLevel(self.logLevel) self.logger.propagate = False self.logger.debug("Constructed") def Init(self, eventCELStarted, celEnvironment): self.eventCELStarted = eventCELStarted self.celEnvironment = celEnvironment def OnCELStarted(self): self.logger.debug("CELStarted!") self.eventCELStarted.set() def OnDataError(self, cqgError, errorDescription): if cqgError is not None: error = win32com.client.Dispatch(cqgError) self.logger.error("OnDataError: Code: %d Description: %s", error.Code, error.Description) self.celEnvironment.SetError() self.eventCELStarted.set() # Class prepares environment for comfortable using of CQG API in python class CELEnvironment: def __init__(self): self.eventCELStarted = threading.Event() self.eventAllPrepared = threading.Event() self.shutdownEvent = win32event.CreateEvent(None, 0, 0, None) self.errorHappened = False self.cqgCEL = None # marshalled in main thread self.cqgCELForCallback = None # marshalled in callback thread self.cqgCELStream = None # stream to marshall CQGCEL into the main thread logging.basicConfig(format=LOG_FORMAT) # set default format in case out logger is inaccessible # set-up CEL logger self.logger = GetLogger() def Init(self, sinkType, customAPIConfiguration): self.thread = Thread(target=self.threadedFunction, args=(sinkType,)) self.thread.start() self.logger.debug("Waiting for CQGCEL creation..") self.eventAllPrepared.wait() self.logger.debug("CQGCEL is created!") self.logger.debug("Marshaled CQGCEL from stream") self.cqgCEL = win32com.client.Dispatch( pythoncom.CoGetInterfaceAndReleaseStream(self.cqgCELStream, pythoncom.IID_IDispatch) ) self.logger.debug("CQGCEL is marshaled from stream!") self.apiConfigurationSet(customAPIConfiguration) self.startupCel() if self.errorHappened: self.logger.fatal("Error happened during CQGCEL start up!") return self.sink def threadedFunction(self, sinkType): self.logger.debug("COM framework is initializing...") pythoncom.CoInitialize() self.logger.info("COM framework is initialized!") self.logger.debug("CQG.CQGCEL.4 creation...") self.cqgCELForCallback = win32com.client.Dispatch("CQG.CQGCEL.4") self.logger.info("CQG.CQGCEL.4 created!") self.logger.debug("Marshal CQGCEL in stream") self.cqgCELStream = pythoncom.CoMarshalInterThreadInterfaceInStream( pythoncom.IID_IDispatch, self.cqgCELForCallback ) self.logger.info("CQGCEL is marshaled in stream") self.logger.debug("Internal sink subscription to events") sinkInternal = win32com.client.WithEvents(self.cqgCELForCallback, SinkInternal) self.logger.info("Internal sink subscribed") self.logger.debug("Sink subscription to events") self.sink = win32com.client.WithEvents(self.cqgCELForCallback, sinkType) self.logger.info("Sink subscribed") sinkInternal.Init(self.eventCELStarted, self) # pylint: disable=no-member self.logger.info("Internal sink is initialized") self.eventAllPrepared.set() self.logger.info("Starting message pumping..") self.pumpMessages() self.logger.debug("COM framework is uninitializing...") pythoncom.CoUninitialize() self.logger.info("COM framework is uninitialized!") def pumpMessages(self): while True: res = win32event.MsgWaitForMultipleObjects( (self.shutdownEvent,), 0, win32event.INFINITE, win32event.QS_ALLEVENTS ) if res == win32event.WAIT_OBJECT_0: break if res == win32event.WAIT_OBJECT_0 + 1: if pythoncom.PumpWaitingMessages(): break # wm_quit else: raise RuntimeError("unexpected win32wait return value") def apiConfigurationSet(self, customAPIConfiguration): self.logger.info("CQGCEL default configuring...") self.cqgCEL.APIConfiguration.CollectionsThrowException = False self.cqgCEL.APIConfiguration.NewInstrumentChangeMode = True self.cqgCEL.APIConfiguration.ReadyStatusCheck = 0 self.cqgCEL.APIConfiguration.TimeZoneCode = constants.tzCentral self.cqgCEL.APIConfiguration.IncludeOrderTransactions = True if customAPIConfiguration is not None: self.logger.info("CQGCEL custom configuring...") customAPIConfiguration(self.cqgCEL.APIConfiguration) def startupCel(self): self.logger.debug("Startup CQGCEL..") self.cqgCEL.Startup() self.logger.debug("Waiting for CQGCEL start..") self.eventCELStarted.wait() self.logger.info("CQGCEL has started.") def Shutdown(self): self.logger.debug("CQGCEL is being shut down..") self.cqgCEL.Shutdown() win32event.SetEvent(self.shutdownEvent) self.thread.join() self.logger.info("CQGCEL is shut down.") def SetError(self): self.errorHappened = True # Function prepares an environment for a sample starting # sampleClassType is a class type (not the instance) that implements the sample # customConfiguration is a function sets APIConfiguration properties in addition to CELEnvironment.apiConfigurationSet def Start(sampleClassType, customConfiguration=None): celEnvironment = CELEnvironment() try: sample = celEnvironment.Init(sampleClassType, customConfiguration) if not celEnvironment.errorHappened: sample.Init(celEnvironment) # pylint: disable=no-member sample.Start() # pylint: disable=no-member except Exception as e: celEnvironment.Shutdown() celEnvironment.logger.fatal("Exception: %s", e) raise e celEnvironment.Shutdown()