"""AlgorithmEngine - Single Source of Truth facade. Usage:: from app.engine import AlgorithmEngine engine = AlgorithmEngine() ctx = engine.analyze(asset_id='bitcoin', timeframe='1h') print(ctx.indicators['rsi_latest']) print(ctx.to_dict()) """ from __future__ import annotations from app.engine.context import AlgorithmContext from app.engine.registry import AlgorithmRegistry class AlgorithmEngine: """Facade for running all algorithms on a asset. This is the Single Source of Truth: every feature (scoring, timeline, analysis pages) accesses algorithms exclusively through this engine. """ def __init__(self): self.registry = AlgorithmRegistry() self._register_defaults() def _register_defaults(self): """Register all default algorithms.""" from app.engine.technical import ALL_TECHNICAL from app.engine.patterns import ALL_PATTERNS from app.engine.quantitative import ALL_QUANTITATIVE from app.engine.ml import ALL_ML from app.engine.strategy import ALL_STRATEGY from app.engine.behavioral import ALL_BEHAVIORAL from app.engine.velocity import ALL_VELOCITY from app.engine.money_management import ALL_MONEY_MGMT for cls in ALL_TECHNICAL: self.registry.register(cls()) for cls in ALL_PATTERNS: self.registry.register(cls()) for cls in ALL_QUANTITATIVE: self.registry.register(cls()) for cls in ALL_ML: self.registry.register(cls()) for cls in ALL_BEHAVIORAL: self.registry.register(cls()) for cls in ALL_VELOCITY: self.registry.register(cls()) for cls in ALL_MONEY_MGMT: self.registry.register(cls()) for cls in ALL_STRATEGY: self.registry.register(cls()) def analyze(self, asset_id: str, timeframe: str = '1h', source: str = None, ohlcv_df: pd.DataFrame = None, batch_scan: bool = False) -> AlgorithmContext: """Run full analysis pipeline on a asset. Args: asset_id: CoinGecko asset ID timeframe: OHLCV timeframe source: Data source (None = use active setting) ohlcv_df: Pre-loaded DataFrame (optional; if None, loads from DB) batch_scan: If True, skip heavy optional steps (Optuna, cross-asset) Returns: Populated AlgorithmContext """ import pandas as pd # noqa: F811 from app.models.settings import AppSettings from app.helpers.asset_filter import get_source_for_coin if source is None: source = get_source_for_coin(asset_id) # Load OHLCV data from database if not provided if ohlcv_df is None: ohlcv_df = self._load_ohlcv(asset_id, timeframe, source) if ohlcv_df is None or ohlcv_df.empty: ctx = AlgorithmContext(asset_id=asset_id, timeframe=timeframe, source=source) ctx.errors.append({'algorithm': 'engine', 'error': 'No OHLCV data available'}) return ctx # Build context ctx = AlgorithmContext( asset_id=asset_id, timeframe=timeframe, source=source, ohlcv=ohlcv_df, current_price=float(ohlcv_df['close'].iloc[-1]), current_volume=float(ohlcv_df['volume'].iloc[-1]), batch_scan=batch_scan, ) # Detect asset type from asset model from app.models.asset import Asset asset = Asset.query.get(asset_id) asset_type = asset.asset_type if asset else 'crypto' # Load settings — mode-aware fees & parameters if asset_type == 'stock': ctx.asset_type = 'stock' ctx.buy_fee_pct = float(AppSettings.get('stock_buy_fee_pct', 0.15)) ctx.sell_fee_pct = float(AppSettings.get('stock_sell_fee_pct', 0.25)) ctx.sell_tax_pct = float(AppSettings.get('stock_sell_tax_pct', 0.10)) ctx.lot_size = asset.lot_size if asset else 100 ctx.market_hours = (9, 16) # IDX: 09:00–16:00 WIB elif asset_type == 'stock_us': ctx.asset_type = 'stock_us' ctx.buy_fee_pct = float(AppSettings.get('stock_us_buy_fee_pct', 0.0)) ctx.sell_fee_pct = float(AppSettings.get('stock_us_sell_fee_pct', 0.0)) ctx.sell_tax_pct = 0.0 ctx.lot_size = asset.lot_size if asset else 1 ctx.market_hours = (9, 16) # NYSE/NASDAQ: 09:30–16:00 ET else: ctx.asset_type = 'crypto' ctx.buy_fee_pct = float(AppSettings.get('buy_fee_pct', 0.31)) ctx.sell_fee_pct = float(AppSettings.get('sell_fee_pct', 0.31)) ctx.sell_tax_pct = 0.0 ctx.lot_size = 0 # fractional ctx.market_hours = () # 24/7 ctx.capital_idr = float(AppSettings.get('default_capital', 1_000_000)) ctx.active_pct = float(AppSettings.get('active_capital_pct', 75)) ctx.reserve_pct = float(AppSettings.get('reserve_capital_pct', 25)) # Run pipeline ctx = self.registry.run_all(ctx) # Free the raw DataFrame reference to reduce memory after pipeline # (algorithms have already extracted what they need into ctx dicts) ctx.ohlcv = None return ctx def _get_ohlcv_limit(self) -> int: """Read OHLCV limit from Flask config (ML_OHLCV_LIMIT).""" try: from flask import current_app return current_app.config.get('ML_OHLCV_LIMIT', 1000) except RuntimeError: return 1000 def _load_ohlcv(self, asset_id: str, timeframe: str, source: str) -> pd.DataFrame | None: """Load OHLCV data from database as DataFrame. Loads only the most recent ML_OHLCV_LIMIT rows (from config) to keep memory usage bounded (~2MB vs potentially 50MB+ for full history). """ import pandas as pd # noqa: F811 from app.models.ohlcv import OHLCVData limit = self._get_ohlcv_limit() # Subquery to get only the latest N rows efficiently records = ( OHLCVData.query .filter_by(asset_id=asset_id, timeframe=timeframe, source=source) .order_by(OHLCVData.datetime_wib.desc()) .limit(limit) .all() ) if not records: return None # Reverse to chronological order (was desc for LIMIT efficiency) records.reverse() data = [] for r in records: data.append({ 'timestamp': r.timestamp, 'datetime': r.datetime_wib, 'open': float(r.open), 'high': float(r.high), 'low': float(r.low), 'close': float(r.close), 'volume': float(r.volume), }) return pd.DataFrame(data) def list_algorithms(self) -> list[dict]: """List all registered algorithms.""" return self.registry.list_algorithms() def get_execution_order(self) -> list[str]: """Get the dependency-resolved execution order.""" return self.registry.get_execution_order()