"""Timeline Generator - creates buy/sell prediction queue for a asset. Generates a daily queue of predicted buy/sell actions with: - Entry/exit prices from ML predictions and technical signals - Net-of-fee profit calculations (Ajaib 0.31%) - Running accumulated units and cash (compounding) - Only generates profitable sell signals (strict validation) Philosophy: "Profit kecil yang sudah terealisasi lebih baik daripada profit besar yang masih berupa harapan." """ from __future__ import annotations from datetime import date, datetime, timedelta from typing import Optional from app.engine.context import AlgorithmContext class TimelineGenerator: """Generates buy/sell prediction queues.""" def __init__(self, engine: Optional[AlgorithmEngine] = None): if engine is None: from app.engine import AlgorithmEngine engine = AlgorithmEngine() self.engine = engine def generate(self, asset_id: str, prediction_date: date = None, timeframe: str = '1h', source: str = None) -> dict: """Generate a full daily prediction timeline. Args: asset_id: CoinGecko asset ID prediction_date: Target date (default: today) timeframe: Analysis timeframe source: Data source Returns: dict with queue entries, summary, and analysis context """ if prediction_date is None: prediction_date = date.today() # Run full analysis pipeline ctx = self.engine.analyze(asset_id, timeframe, source) if ctx.errors and any(e.get('algorithm') == 'engine' for e in ctx.errors): return { 'asset_id': asset_id, 'prediction_date': str(prediction_date), 'queue': [], 'summary': {'error': 'No OHLCV data available'}, 'errors': ctx.errors, } # Generate queue entries queue = self._build_queue(ctx, prediction_date) # Build summary summary = self._build_summary(queue, ctx) return { 'asset_id': asset_id, 'prediction_date': str(prediction_date), 'timeframe': timeframe, 'current_price': ctx.current_price, 'queue': queue, 'summary': summary, 'money_mgmt': ctx.money_mgmt, 'velocity': ctx.velocity, 'errors': ctx.errors, } def _build_queue(self, ctx: AlgorithmContext, pred_date: date) -> list: """Build the prediction queue from analysis context.""" queue = [] seq = 0 price = ctx.current_price buy_fee_pct = ctx.buy_fee_pct / 100 sell_fee_pct = ctx.sell_fee_pct / 100 # Capital setup ff = ctx.money_mgmt.get('fixed_fractional', {}) active_capital = ff.get('active_capital', ctx.capital_idr * ctx.active_pct / 100) # Position sizer results ps = ctx.money_mgmt.get('position_sizer', {}) tp_levels = ps.get('take_profit', {}) sl_price = ps.get('stop_loss', price * 0.985) # Running state for compounding running_cash = active_capital running_units = 0.0 # === Entry 1: Initial BUY (primary signal) === overall_signal = self._determine_signal(ctx) if overall_signal in ('BUY', 'HOLD'): entry_price = price position_value = min(running_cash * 0.25, running_cash) buy_fee = position_value * buy_fee_pct net_buy = position_value - buy_fee units = net_buy / entry_price if entry_price > 0 else 0 seq += 1 queue.append({ 'sequence_num': seq, 'action': 'BUY', 'predicted_price': ctx.round_price(entry_price), 'quantity': float(units), 'cash_amount': round(position_value, 2), 'fee_amount': round(buy_fee, 2), 'profit': None, 'accumulated_units': float(running_units + units), 'accumulated_cash': round(running_cash - position_value, 2), 'confidence': self._calc_confidence(ctx), 'reason': self._entry_reason(ctx), 'algorithm_used': list(ctx.signal_contributions.keys()), }) running_cash -= position_value running_units += units entry_cost = position_value # track for profit calc # === Entry 2: Take Profit 1 (1.2x R:R) === tp1 = tp_levels.get('tp1', price * 1.010) sell_units_tp1 = running_units * 0.4 # sell 40% at TP1 if sell_units_tp1 > 0 and tp1 > entry_price: gross_revenue = sell_units_tp1 * tp1 sell_fee = gross_revenue * sell_fee_pct net_revenue = gross_revenue - sell_fee cost_basis = sell_units_tp1 * entry_price * (1 + buy_fee_pct) profit = net_revenue - cost_basis # Must exceed total fees by 20% (not just > 0) fee_threshold = cost_basis * (buy_fee_pct + sell_fee_pct) * 1.2 if profit > fee_threshold: seq += 1 running_units -= sell_units_tp1 running_cash += net_revenue queue.append({ 'sequence_num': seq, 'action': 'SELL', 'predicted_price': ctx.round_price(tp1, 'up'), 'quantity': float(sell_units_tp1), 'cash_amount': round(net_revenue, 2), 'fee_amount': round(sell_fee, 2), 'profit': round(profit, 2), 'accumulated_units': float(running_units), 'accumulated_cash': round(running_cash, 2), 'confidence': self._calc_confidence(ctx) * 0.9, 'reason': f'Take Profit 1 (1.2x R:R) at Rp {tp1:,.0f}', 'algorithm_used': ['money_mgmt.position_sizer'], }) # === Entry 3: Take Profit 2 (1.8x R:R) === tp2 = tp_levels.get('tp2', price * 1.018) sell_units_tp2 = running_units * 0.5 # sell 50% remaining at TP2 if sell_units_tp2 > 0 and tp2 > entry_price: gross_revenue = sell_units_tp2 * tp2 sell_fee = gross_revenue * sell_fee_pct net_revenue = gross_revenue - sell_fee cost_basis = sell_units_tp2 * entry_price * (1 + buy_fee_pct) profit = net_revenue - cost_basis fee_threshold = cost_basis * (buy_fee_pct + sell_fee_pct) * 1.2 if profit > fee_threshold: seq += 1 running_units -= sell_units_tp2 running_cash += net_revenue queue.append({ 'sequence_num': seq, 'action': 'SELL', 'predicted_price': ctx.round_price(tp2, 'up'), 'quantity': float(sell_units_tp2), 'cash_amount': round(net_revenue, 2), 'fee_amount': round(sell_fee, 2), 'profit': round(profit, 2), 'accumulated_units': float(running_units), 'accumulated_cash': round(running_cash, 2), 'confidence': self._calc_confidence(ctx) * 0.8, 'reason': f'Take Profit 2 (1.8x R:R) at Rp {tp2:,.0f}', 'algorithm_used': ['money_mgmt.position_sizer'], }) # === Entry 4: Re-entry BUY (if mean reversion or ladder) === mr = ctx.velocity.get('mean_reversion', {}) if mr.get('signal') == 'OVERSOLD' and running_cash > position_value * 0.5: reentry_price = mr.get('reversion_target', price * 0.99) if reentry_price < price: # only buy lower rebuy_value = min(running_cash * 0.20, running_cash) rebuy_fee = rebuy_value * buy_fee_pct rebuy_units = (rebuy_value - rebuy_fee) / reentry_price if reentry_price > 0 else 0 seq += 1 running_cash -= rebuy_value running_units += rebuy_units queue.append({ 'sequence_num': seq, 'action': 'BUY', 'predicted_price': ctx.round_price(reentry_price, 'down'), 'quantity': float(rebuy_units), 'cash_amount': round(rebuy_value, 2), 'fee_amount': round(rebuy_fee, 2), 'profit': None, 'accumulated_units': float(running_units), 'accumulated_cash': round(running_cash, 2), 'confidence': self._calc_confidence(ctx) * 0.7, 'reason': f'Mean reversion re-entry (z={mr.get("zscore", 0):.1f})', 'algorithm_used': ['velocity.mean_reversion'], }) # === Entry 5: Take Profit 3 / Trail (2.5x R:R) === tp3 = tp_levels.get('tp3', price * 1.025) if running_units > 0 and tp3 > entry_price: gross_revenue = running_units * tp3 sell_fee = gross_revenue * sell_fee_pct net_revenue = gross_revenue - sell_fee cost_basis = running_units * entry_price * (1 + buy_fee_pct) profit = net_revenue - cost_basis fee_threshold = cost_basis * (buy_fee_pct + sell_fee_pct) * 1.2 if profit > fee_threshold: seq += 1 final_units = running_units running_units = 0 running_cash += net_revenue queue.append({ 'sequence_num': seq, 'action': 'SELL', 'predicted_price': ctx.round_price(tp3, 'up'), 'quantity': float(final_units), 'cash_amount': round(net_revenue, 2), 'fee_amount': round(sell_fee, 2), 'profit': round(profit, 2), 'accumulated_units': float(running_units), 'accumulated_cash': round(running_cash, 2), 'confidence': self._calc_confidence(ctx) * 0.6, 'reason': f'Take Profit 3 (2.5x R:R) at Rp {tp3:,.0f} - close all', 'algorithm_used': ['money_mgmt.position_sizer'], }) elif overall_signal == 'SELL': # Pure sell scenario - exit existing position seq += 1 queue.append({ 'sequence_num': seq, 'action': 'SELL', 'predicted_price': ctx.round_price(price), 'quantity': 0, 'cash_amount': 0, 'fee_amount': 0, 'profit': None, 'accumulated_units': 0, 'accumulated_cash': round(running_cash, 2), 'confidence': self._calc_confidence(ctx), 'reason': 'SELL signal - exit or reduce existing position', 'algorithm_used': list(ctx.signal_contributions.keys()), }) # === Stop Loss entry (always present for BUY queues) === if len(queue) > 0 and queue[0]['action'] == 'BUY' and running_units > 0: seq += 1 sl = ps.get('stop_loss', price * 0.985) sl_revenue = running_units * sl * (1 - sell_fee_pct) sl_cost = running_units * price * (1 + buy_fee_pct) queue.append({ 'sequence_num': seq, 'action': 'SELL', 'predicted_price': ctx.round_price(sl, 'down'), 'quantity': float(running_units), 'cash_amount': round(sl_revenue, 2), 'fee_amount': round(running_units * sl * sell_fee_pct, 2), 'profit': round(sl_revenue - sl_cost, 2), 'accumulated_units': 0, 'accumulated_cash': round(running_cash + sl_revenue, 2), 'confidence': 0, 'reason': f'STOP LOSS at Rp {sl:,.0f} (contingency)', 'algorithm_used': ['money_mgmt.position_sizer'], }) return queue def _determine_signal(self, ctx: AlgorithmContext) -> str: """Determine overall signal from contributions.""" buy_w = 0 sell_w = 0 hold_w = 0 for _, contrib in ctx.signal_contributions.items(): sig = contrib.get('signal', 'HOLD') w = contrib.get('weight', 0) if sig == 'BUY': buy_w += w elif sig == 'SELL': sell_w += w else: hold_w += w total = buy_w + sell_w + hold_w if total == 0: return 'HOLD' score = (buy_w - sell_w) / total * 100 if score > 30: return 'BUY' elif score < -30: return 'SELL' return 'HOLD' def _calc_confidence(self, ctx: AlgorithmContext) -> float: """Calculate confidence score 0-100.""" contribs = ctx.signal_contributions if not contribs: return 0 buy_w = sum(c.get('weight', 0) for c in contribs.values() if c.get('signal') == 'BUY') sell_w = sum(c.get('weight', 0) for c in contribs.values() if c.get('signal') == 'SELL') total = sum(c.get('weight', 0) for c in contribs.values()) if total == 0: return 0 agreement = abs(buy_w - sell_w) / total error_penalty = len(ctx.errors) * 5 return round(min(100, max(0, agreement * 100 - error_penalty)), 1) def _entry_reason(self, ctx: AlgorithmContext) -> str: """Build human-readable entry reason from top contributors.""" buy_reasons = [] for aid, contrib in ctx.signal_contributions.items(): if contrib.get('signal') == 'BUY': buy_reasons.append(contrib.get('reason', aid)) if buy_reasons: return '; '.join(buy_reasons[:3]) return 'Multiple indicators suggest entry' def save_queue(self, timeline_result: dict) -> int: """Save the generated queue to database. Returns: Number of entries saved """ from app.models.prediction_queue import PredictionQueue from app.extensions import db from datetime import date as date_type asset_id = timeline_result['asset_id'] pred_date_str = timeline_result['prediction_date'] pred_date = date_type.fromisoformat(pred_date_str) # Delete existing entries for this asset/date PredictionQueue.query.filter_by( asset_id=asset_id, prediction_date=pred_date, ).delete() count = 0 for entry in timeline_result.get('queue', []): pq = PredictionQueue( asset_id=asset_id, prediction_date=pred_date, sequence_num=entry['sequence_num'], action=entry['action'], predicted_price=entry['predicted_price'], quantity=entry['quantity'], cash_amount=entry['cash_amount'], fee_amount=entry['fee_amount'], profit=entry.get('profit'), accumulated_units=entry.get('accumulated_units'), accumulated_cash=entry.get('accumulated_cash'), confidence=entry.get('confidence'), reason=entry.get('reason'), algorithm_used=entry.get('algorithm_used'), ) db.session.add(pq) count += 1 db.session.commit() return count def _build_summary(self, queue: list, ctx: AlgorithmContext) -> dict: """Build summary of the prediction queue.""" if not queue: return { 'total_entries': 0, 'signal': 'NO_DATA', } buys = [e for e in queue if e['action'] == 'BUY'] sells = [e for e in queue if e['action'] == 'SELL' and (e.get('profit') or 0) > 0] stop_losses = [e for e in queue if 'STOP LOSS' in (e.get('reason') or '')] total_invested = sum(e['cash_amount'] for e in buys) total_profit = sum(e.get('profit') or 0 for e in sells) total_fees = sum(e['fee_amount'] for e in queue) # Net return net_return_pct = (total_profit / total_invested * 100) if total_invested > 0 else 0 return { 'total_entries': len(queue), 'num_buys': len(buys), 'num_sells': len(sells), 'num_stop_losses': len(stop_losses), 'total_invested_idr': round(total_invested, 0), 'total_profit_idr': round(total_profit, 0), 'total_fees_idr': round(total_fees, 0), 'net_return_pct': round(net_return_pct, 2), 'signal': queue[0]['action'] if queue else 'HOLD', 'confidence': queue[0].get('confidence', 0) if queue else 0, }