-
Notifications
You must be signed in to change notification settings - Fork 1
TVEngine API
- Introduction
- Core Architecture
- Singleton Initialization
- Configuration Management
- Module Loading System
- Indicator Lifecycle Control
- Remote IPC Interface
- Runtime Execution Flow
- Thread Safety and EventBus Integration
- Practical Usage Examples
The TVEngine class serves as the core singleton engine of PyTradingView, orchestrating indicator management, configuration, and execution within the TradingView widget environment. Built with a modular inheritance architecture, TVEngine integrates multiple functional mixins to provide comprehensive capabilities for indicator loading, activation, configuration updates, and remote control via Electron IPC. The engine follows a strict singleton pattern with thread-safe initialization, ensuring exactly one instance exists throughout the application lifecycle. It interacts with the EventBus system for event-driven communication and manages multi-chart contexts for synchronized indicator operations across different layouts.
TVEngine employs a layered inheritance structure where each mixin contributes specific functionality, creating a cohesive yet modular system. The inheritance hierarchy flows from base singleton management up to runtime execution, with TVEngine as the final composite class. This design enables separation of concerns while maintaining a unified interface.
classDiagram
class TVEngineSingleton {
+_instance : Optional[TVEngineSingleton]
+_lock : Optional[threading.Lock]
+__new__(cls, config) TVEngineSingleton
+get_instance(config) TVEngineSingleton
+reset() void
}
class TVEngineLoader {
+load_indicator_from_file(file_path) bool
+load_indicators_from_directory(directory, recursive) int
}
class TVEngineManager {
+activate_indicator(name, chart_id) bool
+deactivate_indicator(name, chart_id) bool
+deactivate_all(chart_id) void
+get_active_indicators(chart_id) Dict[str, TVIndicator]
+get_all_chart_indicators() Dict[str, Dict[str, TVIndicator]]
}
class TVEngineConfig {
+update_config(config_dict) TVEngineConfig
+get_config() Dict[str, Any]
+set_config(config) TVEngineConfig
+get_indicator_config(name, chart_id) Optional[Dict[str, Any]]
+update_indicator_config(name, config_dict, chart_id) Tuple[bool, List[str]]
+update_indicator_input(name, input_id, value, chart_id) Tuple[bool, Optional[str]]
+update_indicator_style(name, style_id, chart_id, **kwargs) Tuple[bool, Optional[str]]
+recalculate_indicator(name, chart_id) Tuple[bool, Optional[str]]
+recalculate_all_indicators(chart_id) Dict[str, Tuple[bool, Optional[str]]]
+get_all_configs(chart_id) Dict[str, Dict[str, Any]]
}
class TVEngineRemote {
+remote_get_status(chart_id) Dict[str, Any]
+remote_update_config(config_dict) Tuple[bool, Optional[str]]
+remote_load_indicator(file_path) Tuple[bool, Optional[str]]
+remote_activate_indicator(name, chart_id) Tuple[bool, Optional[str]]
+remote_deactivate_indicator(name, chart_id) Tuple[bool, Optional[str]]
+remote_update_indicator_config(name, config_dict, chart_id) Tuple[bool, Optional[str]]
+remote_recalculate_indicator(name, chart_id) Tuple[bool, Optional[str]]
+remote_get_indicator_info() Dict[str, Dict[str, Any]]
}
class TVEngineDrawing {
+run_indicators(df, chart_id) Dict[str, Any]
+run_indicators_for_chart(chart_id, df) Dict[str, Any]
+_draw_indicator(indicator, chart, df, signals, drawables) None
+_default_draw(indicator, chart, df, signals, drawables) None
+_draw_signal(indicator, chart, timestamps, signal) None
+_draw_drawable(indicator, chart, timestamps, drawable) None
+_convert_timestamp_to_seconds(timestamp) int
}
class TVEngineRuntime {
+__init__(config) None
+setup(indicators_dir, auto_activate, config) TVEngineRuntime
+run(widget_config, indicators_dir, on_port) None
+_on_chart_ready_event(event) None
+_on_chart_data_ready(widget) None
+_subscribe_layout_changed_event(widget) None
+_handle_layout_changed(widget) None
+_initialize_all_charts(widget) None
+_activate_default_indicators_for_chart(chart_id) None
+_cleanup_all_charts() None
+_setup_chart_data_listener(widget, chart, chart_id, chart_index) None
}
TVEngineSingleton <|-- TVEngineLoader
TVEngineLoader <|-- TVEngineManager
TVEngineManager <|-- TVEngineConfig
TVEngineConfig <|-- TVEngineRemote
TVEngineRemote <|-- TVEngineDrawing
TVEngineDrawing <|-- TVEngineRuntime
TVEngineRuntime <|-- TVEngine
TVEngine implements a thread-safe singleton pattern through the TVEngineSingleton base class, ensuring exactly one instance exists across the application. The implementation uses a class-level lock to synchronize instance creation, preventing race conditions during initialization. The get_instance() class method serves as the primary access point, creating the instance on first call and returning the existing instance thereafter. The singleton maintains an _initialized flag to prevent re-initialization and supports a reset mechanism (intended for testing) that deactivates all indicators before clearing the instance reference.
sequenceDiagram
participant Client
participant TVEngine
participant Lock
Client->>TVEngine : get_instance(config)
TVEngine->>TVEngine : Check if _instance exists
alt Instance does not exist
TVEngine->>Lock : Acquire threading.Lock
TVEngine->>TVEngine : Create new instance
TVEngine->>TVEngine : Set _initialized = True
TVEngine->>Lock : Release lock
TVEngine-->>Client : Return new instance
else Instance exists
TVEngine-->>Client : Return existing instance
end
The TVEngine provides comprehensive configuration management through the TVEngineConfig mixin, enabling dynamic updates to both engine-wide and indicator-specific settings. The configuration system centers around a TVWidgetConfig object that stores widget parameters and can be updated at runtime. Engine configuration updates are performed via update_config() and set_config() methods, which merge new settings into the existing configuration. For indicators, the engine offers granular control through methods that update input parameters, visual styles, and complete configuration dictionaries, with support for both single-chart and global updates.
flowchart TD
Start([Update Configuration]) --> EngineConfig{"Engine or Indicator?"}
EngineConfig --> |Engine| UpdateEngine["update_config(config_dict)"]
UpdateEngine --> Validate["Validate configuration structure"]
Validate --> Update["Merge into self.config"]
Update --> Log["Log update: Engine config updated"]
UpdateEngine --> ReturnSelf["Return self for chaining"]
EngineConfig --> |Indicator| UpdateIndicator{"Single chart or all?"}
UpdateIndicator --> |Specific Chart| UpdateSingle["_update_indicator_config_for_chart()"]
UpdateSingle --> CheckContext["Verify chart context exists"]
CheckContext --> CheckActive["Verify indicator is active"]
CheckActive --> ExecuteUpdate["Call indicator.update_config()"]
ExecuteUpdate --> HandleResult["Return (success, errors)"]
UpdateIndicator --> |All Charts| UpdateAll["Loop through all chart contexts"]
UpdateAll --> EachChart["Call _update_indicator_config_for_chart()"]
EachChart --> CollectResults["Aggregate success status and errors"]
CollectResults --> ReturnResults["Return global (success, all_errors)"]
ReturnSelf --> End([Configuration Updated])
HandleResult --> End
ReturnResults --> End
The module loading system in TVEngine, implemented through TVEngineLoader, enables dynamic discovery and registration of indicator modules from both individual files and directories. The system uses Python's importlib to programmatically load modules, scanning for subclasses of TVIndicator within each loaded file. When loading from a directory, it supports recursive traversal to find indicators in subdirectories. The loader implements a sophisticated registration logic that prevents duplicate registration by checking both class name and decorator-based registration, ensuring indicators are registered exactly once. For each discovered indicator, the system reads its enabled status from its configuration to determine initial activation state.
flowchart TD
Start([Load Indicators]) --> Source{"From file or directory?"}
Source --> |Single File| LoadFile["load_indicator_from_file(file_path)"]
LoadFile --> CheckExists["Verify file exists"]
CheckExists --> LoadModule["Use importlib to load module"]
LoadModule --> FindClasses["Scan for TVIndicator subclasses"]
FindClasses --> CheckFound{"Indicators found?"}
CheckFound --> |No| ReturnFalse["Return False"]
CheckFound --> |Yes| RegisterLoop["For each indicator class"]
RegisterLoop --> CheckRegistered{"Already registered?"}
CheckRegistered --> |Yes| SkipRegistration["Skip auto-registration"]
CheckRegistered --> |No| ReadConfig["Read enabled status from config"]
ReadConfig --> Register["Register with IndicatorRegistry"]
Register --> LogSuccess["Log: Loaded indicator (enabled=X)"]
Source --> |Directory| LoadDir["load_indicators_from_directory(directory)"]
LoadDir --> CheckDir["Verify directory exists"]
CheckDir --> FindFiles["Glob for *.py files"]
FindFiles --> FilterFiles["Skip __init__.py and __pycache__"]
FilterFiles --> ProcessFiles["For each file_path"]
ProcessFiles --> CallLoadFile["load_indicator_from_file(file_path)"]
CallLoadFile --> CountSuccess["Increment loaded_count"]
SkipRegistration --> NextClass
Register --> NextClass
NextClass{"More classes?"} --> |Yes| RegisterLoop
NextClass --> |No| ReturnTrue["Return True"]
CountSuccess --> MoreFiles{"More files?"} --> |Yes| ProcessFiles
MoreFiles --> |No| LogSummary["Log: Loaded X indicators from directory"]
LogSummary --> ReturnCount["Return loaded_count"]
ReturnFalse --> End([Loading Complete])
ReturnTrue --> End
ReturnCount --> End
Indicator lifecycle management in TVEngine is handled by the TVEngineManager mixin, which provides methods to activate, deactivate, and query the status of indicators across one or multiple charts. Activation creates a new indicator instance from the registry, associates it with a chart context, and optionally initializes it immediately if the widget and chart are available. Deactivation triggers the indicator's destroy callback to clean up resources before removing it from the context. The system maintains indicator state through ChartContextManager, which tracks active indicators per chart and supports operations on all charts simultaneously when no specific chart ID is provided.
sequenceDiagram
participant Client
participant TVEngine
participant Registry
participant Context
participant Indicator
Client->>TVEngine : activate_indicator("FalseBreakout", "chart_0")
TVEngine->>TVEngine : Check chart_id provided
TVEngine->>TVEngine : Call _activate_indicator_to_chart()
TVEngine->>Context : Get context for "chart_0"
Context-->>TVEngine : Return context
TVEngine->>Context : Check if indicator already active
TVEngine->>Registry : create_instance("FalseBreakout")
Registry-->>TVEngine : Return indicator instance
TVEngine->>Indicator : set_chart_id("chart_0")
TVEngine->>Indicator : on_init(widget, chart, chart_id) if available
TVEngine->>Context : add_indicator("FalseBreakout", indicator)
TVEngine-->>Client : Return True
Client->>TVEngine : deactivate_indicator("FalseBreakout", "chart_0")
TVEngine->>TVEngine : Call _deactivate_indicator_from_chart()
TVEngine->>Context : Get context for "chart_0"
Context-->>TVEngine : Return context
TVEngine->>Context : Check if indicator active
TVEngine->>Indicator : on_destroy()
TVEngine->>Context : remove_indicator("FalseBreakout")
TVEngine-->>Client : Return True
The TVEngineRemote mixin exposes a comprehensive set of methods for external control of the engine via Electron IPC, enabling remote clients to query status, update configurations, and manage indicators. Each remote method follows a consistent pattern of returning a success flag and optional error message, wrapping the corresponding internal method call in a try-except block to handle exceptions gracefully. The remote interface provides complete control over the engine's functionality, including getting detailed status information, loading new indicators, activating/deactivating indicators, and updating their configurations, making it suitable for integration with external UIs or control systems.
flowchart TD
Client[Remote Client] --> Bridge[Electron IPC Bridge]
Bridge --> Engine[TVEngine Remote Interface]
subgraph TVEngine Remote Methods
Engine --> GetStatus["remote_get_status(chart_id)"]
Engine --> UpdateConfig["remote_update_config(config_dict)"]
Engine --> LoadIndicator["remote_load_indicator(file_path)"]
Engine --> ActivateIndicator["remote_activate_indicator(name, chart_id)"]
Engine --> DeactivateIndicator["remote_deactivate_indicator(name, chart_id)"]
Engine --> UpdateIndicatorConfig["remote_update_indicator_config(name, config_dict, chart_id)"]
Engine --> RecalculateIndicator["remote_recalculate_indicator(name, chart_id)"]
Engine --> GetIndicatorInfo["remote_get_indicator_info()"]
end
GetStatus --> |Success| StatusResponse["{initialized, config, all_indicators, enabled_indicators, widget_connected, chart_count, chart_ids, charts}"]
GetStatus --> |Error| Error404["{error: 'Chart not found'}"]
UpdateConfig --> TryBlock1["try:"]
TryBlock1 --> CallUpdate["self.update_config(config_dict)"]
CallUpdate --> ReturnSuccess1["Return (True, None)"]
TryBlock1 --> Except1["except Exception as e:"]
Except1 --> LogError1["logger.error(f'Failed to update config: {str(e)}')"]
Except1 --> ReturnError1["Return (False, error_msg)"]
LoadIndicator --> TryBlock2["try:"]
TryBlock2 --> CallLoad["self.load_indicator_from_file(file_path)"]
CallLoad --> CheckSuccess["if success:"]
CheckSuccess --> ReturnSuccess2["Return (True, None)"]
CheckSuccess --> ReturnFail["Return (False, 'Failed to load')"]
TryBlock2 --> Except2["except Exception as e:"]
Except2 --> LogError2["logger.error(f'Exception while loading: {str(e)}', exc_info=True)"]
Except2 --> ReturnError2["Return (False, error_msg)"]
ActivateIndicator --> TryBlock3["try:"]
TryBlock3 --> CallActivate["self.activate_indicator(name, chart_id)"]
CallActivate --> CheckSuccess3["if success:"]
CheckSuccess3 --> ReturnSuccess3["Return (True, None)"]
CheckSuccess3 --> ReturnFail3["Return (False, 'Failed to activate')"]
TryBlock3 --> Except3["except Exception as e:"]
Except3 --> LogError3["logger.error(f'Exception while activating: {str(e)}', exc_info=True)"]
Except3 --> ReturnError3["Return (False, error_msg)"]
StatusResponse --> Bridge
Error404 --> Bridge
ReturnSuccess1 --> Bridge
ReturnError1 --> Bridge
ReturnSuccess2 --> Bridge
ReturnFail --> Bridge
ReturnError2 --> Bridge
ReturnSuccess3 --> Bridge
ReturnFail3 --> Bridge
ReturnError3 --> Bridge
Bridge --> Client
The TVEngineRuntime mixin defines the core execution flow of the engine, from initialization through setup to the main run loop. The process begins with configuration handling and indicator loading, followed by registration of widget callbacks and event listeners. The engine subscribes to the CHART_READY event to initialize charts when they become available. It also listens for layout changes, triggering a complete reinitialization of all charts when the user switches layouts. For each chart, the engine sets up a data loading listener that triggers indicator calculations and drawing when new data arrives, creating a responsive and dynamic trading environment.
flowchart TD
Start([TVEngine.run()]) --> HandleConfig["Handle Configuration"]
HandleConfig --> MergeConfig["Merge runtime widget_config"]
MergeConfig --> ValidateConfig["Validate configuration"]
ValidateConfig --> LoadIndicators["Load indicators from indicators_dir"]
LoadIndicators --> SetupCallbacks["Setup Widget Callbacks"]
SetupCallbacks --> RegisterConfig["Register config provider with TVBridge"]
RegisterConfig --> RegisterChartReady["Register chart ready callback"]
RegisterChartReady --> SetupEvents["Setup Event Listeners"]
SetupEvents --> SubscribeChartReady["Subscribe to CHART_READY event"]
SubscribeChartReady --> StartService["Start Bridge Service"]
StartService --> LogStart["Log: Starting Indicator Engine..."]
LogStart --> PublishEvent["Publish BRIDGE_STARTED event"]
PublishEvent --> RunBridge["bridge.run(on_port)"]
RunBridge --> ChartReadyEvent(CHART_READY Event)
ChartReadyEvent --> OnChartReady["_on_chart_data_ready(widget)"]
OnChartReady --> SaveWidget["Save widget reference"]
SaveWidget --> SubscribeLayout["Subscribe to layout_changed event"]
SubscribeLayout --> InitializeCharts["_initialize_all_charts(widget)"]
InitializeCharts --> GetCount["Get chartsCount()"]
GetCount --> ForEachChart["For each chart_index"]
ForEachChart --> CreateContext["Create chart context"]
CreateContext --> ActivateDefaults["Activate default indicators"]
ActivateDefaults --> SetupDataListener["Setup onDataLoaded listener"]
SetupDataListener --> DataLoadedEvent(onDataLoaded Event)
DataLoadedEvent --> OnDataLoaded["on_data_loaded_callback()"]
OnDataLoaded --> GetContext["Get chart context"]
GetContext --> InitIndicators["Initialize active indicators"]
InitIndicators --> ExportData["chart.exportData(callback=export_callback)"]
ExportData --> ExportCallback(export_callback)
ExportCallback --> ConvertData["Convert to DataFrame"]
ConvertData --> RunIndicators["run_indicators_for_chart(chart_id, df)"]
RunIndicators --> DrawResults["Draw signals and drawables"]
DrawResults --> LogSummary["Log result summary"]
SubscribeLayout --> LayoutChangedEvent(layout_changed Event)
LayoutChangedEvent --> OnLayoutChanged["_handle_layout_changed(widget)"]
OnLayoutChanged --> LogReinit["Log: Reinitializing charts..."]
OnLayoutChanged --> CleanupCharts["_cleanup_all_charts()"]
CleanupCharts --> Reinitialize["_initialize_all_charts(widget)"]
Reinitialize --> Complete["Chart reinitialization completed"]
TVEngine ensures thread safety through a combination of the singleton pattern's locking mechanism and the EventBus system for synchronized event communication. The singleton's __new__ method uses a threading.Lock to prevent race conditions during instance creation, while the EventBus provides a thread-safe publish-subscribe mechanism for inter-component communication. The engine subscribes to critical events like CHART_READY and BRIDGE_STARTED, allowing components to react to system state changes in a coordinated manner. This event-driven architecture decouples components and ensures that operations like chart initialization and layout changes are handled consistently across all parts of the system.
The following examples demonstrate common usage patterns for the TVEngine API, showing how to initialize the engine, configure it, load indicators, and manage them remotely.
# Example 1: Basic Engine Setup and Execution
engine = TVEngine.get_instance()
engine.setup('./indicators', config={'symbol': 'BTCUSDT'})
engine.run()
# Example 2: Configuration Management
engine.update_config({'interval': '1h', 'exchange': 'Binance'})
current_config = engine.get_config()
# Example 3: Module Loading
engine.load_indicator_from_file('./indicators/custom_indicator.py')
engine.load_indicators_from_directory('./indicators', recursive=True)
# Example 4: Indicator Lifecycle Control
engine.activate_indicator('FalseBreakout', 'chart_0')
engine.deactivate_indicator('FalseBreakout', 'chart_0')
# Example 5: Remote IPC Operations
status = engine.remote_get_status()
success, error = engine.remote_activate_indicator('FalseBreakout')
success, error = engine.remote_update_indicator_config('FalseBreakout', {'length': 20})