3.9 Enhance chart functionality with indicator management and data export features

- Updated `ChartBuilder` to support dynamic indicator integration, allowing users to specify overlay and subplot indicators for enhanced chart analysis.
- Implemented a new `get_indicator_data` method in `MarketDataIntegrator` for fetching indicator data based on user configurations.
- Added `create_export_controls` in `chart_controls.py` to facilitate data export options (CSV/JSON) for user analysis.
- Enhanced error handling and logging throughout the chart and data analysis processes to improve reliability and user feedback.
- Updated documentation to reflect new features and usage guidelines for indicator management and data export functionalities.
This commit is contained in:
Vasily.onl
2025-06-06 12:57:35 +08:00
parent 8572a7a387
commit c121b469f0
10 changed files with 512 additions and 654 deletions

View File

@@ -415,6 +415,67 @@ class TechnicalIndicators:
return results
def calculate(self, indicator_type: str, candles: Union[pd.DataFrame, List[OHLCVCandle]], **kwargs) -> Optional[Dict[str, Any]]:
"""
Generic method to calculate any supported indicator by type.
Args:
indicator_type: The type of indicator to calculate (e.g., 'sma', 'ema').
candles: The input data, either a DataFrame or a list of OHLCVCandle objects.
**kwargs: Keyword arguments for the specific indicator function.
Returns:
A dictionary containing the indicator results, or None if the type is unknown.
"""
# If input is a DataFrame, convert it to list of OHLCVCandle objects.
# This is a temporary adaptation to the existing methods.
# Future optimization should standardize on DataFrames.
if isinstance(candles, pd.DataFrame):
from .data_types import OHLCVCandle
# Ensure required columns are present
required_cols = {'open', 'high', 'low', 'close', 'volume'}
if not required_cols.issubset(candles.columns):
if self.logger:
self.logger.error("Indicators: DataFrame missing required columns for OHLCVCandle conversion.")
return None
symbol = kwargs.get('symbol', 'UNKNOWN')
timeframe = kwargs.get('timeframe', 'UNKNOWN')
candles_list = [
OHLCVCandle(
symbol=symbol,
timeframe=timeframe,
start_time=row['timestamp'],
end_time=row['timestamp'],
open=Decimal(str(row['open'])),
high=Decimal(str(row['high'])),
low=Decimal(str(row['low'])),
close=Decimal(str(row['close'])),
volume=Decimal(str(row['volume'])),
trade_count=int(row.get('trade_count', 0))
) for _, row in candles.iterrows()
]
candles = candles_list
indicator_method = getattr(self, indicator_type, None)
if indicator_method and callable(indicator_method):
# We need to construct a proper IndicatorResult object here
# For now, let's adapt to what the methods return
raw_result = indicator_method(candles, **kwargs)
# The methods return List[IndicatorResult], let's package that
if raw_result:
return {
"data": raw_result
}
return None
if self.logger:
self.logger.warning(f"TechnicalIndicators: Unknown indicator type '{indicator_type}'")
return None
def create_default_indicators_config() -> Dict[str, Dict[str, Any]]:
"""