from datetime import datetime from enum import Enum from typing import Optional, Dict, Any, List from dataclasses import dataclass from db.db_connection import create_client class PlanStatus(Enum): ACTIVE = 'active' ARCHIVED = 'archived' TESTING = 'testing' DEPRECATED = 'deprecated' class Timeframe(Enum): DAILY = 'daily' WEEKLY = 'weekly' HOURLY = 'hourly' MIN_15 = '15-min' MIN_30 = '30-min' MIN_5 = '5-min' class MarketFocus(Enum): STOCKS = 'stocks' CRYPTO = 'crypto' FOREX = 'forex' OPTIONS = 'options' FUTURES = 'futures' class TradeFrequency(Enum): DAILY = 'daily' WEEKLY = 'weekly' MONTHLY = 'monthly' AS_NEEDED = 'as-needed' @dataclass class TradingPlan: # General Info plan_name: str status: PlanStatus timeframe: Timeframe market_focus: MarketFocus entry_criteria: str exit_criteria: str stop_loss: float profit_target: float risk_reward_ratio: float trade_frequency: TradeFrequency market_conditions: str indicators_used: str entry_confirmation: str position_sizing: float maximum_drawdown: float max_trades_per_day: int max_trades_per_week: int total_risk_per_trade: float max_portfolio_risk: float adjustments_for_drawdown: str risk_controls: str plan_author: str # Fields with default values must come after required fields created_at: datetime = None updated_at: datetime = None id: int = None strategy_version: int = 1 win_rate: Optional[float] = None average_return_per_trade: Optional[float] = None profit_factor: Optional[float] = None historical_backtest_results: Optional[str] = None real_trade_performance: Optional[str] = None improvements_needed: Optional[str] = None trade_review_notes: Optional[str] = None future_testing_ideas: Optional[str] = None sector_focus: Optional[str] = None fundamental_criteria: Optional[str] = None options_strategy_details: Optional[str] = None def create_trading_plan_table(): """Create the trading plans table if it doesn't exist""" with create_client() as client: try: # Create new table with a structure that supports updates query = """ CREATE TABLE IF NOT EXISTS trading_plans ( id UInt32, plan_name String, status String, created_at DateTime, updated_at DateTime, timeframe String, market_focus String, entry_criteria String, exit_criteria String, stop_loss Float64, profit_target Float64, risk_reward_ratio Float64, trade_frequency String, market_conditions String, indicators_used String, entry_confirmation String, position_sizing Float64, maximum_drawdown Float64, max_trades_per_day UInt32, max_trades_per_week UInt32, total_risk_per_trade Float64, max_portfolio_risk Float64, adjustments_for_drawdown String, risk_controls String, win_rate Float64, average_return_per_trade Float64, profit_factor Float64, historical_backtest_results String, real_trade_performance String, improvements_needed String, strategy_version UInt32, plan_author String, trade_review_notes String, future_testing_ideas String, sector_focus String, fundamental_criteria String, options_strategy_details String ) ENGINE = ReplacingMergeTree() ORDER BY id """ client.command(query) print("Table 'trading_plans' created successfully.") except Exception as e: print(f"Error creating table: {e}") def save_trading_plan(plan: TradingPlan) -> int: """Save a trading plan to the database""" with create_client() as client: if not plan.id: # Generate new ID for new plans result = client.query("SELECT max(id) FROM trading_plans") max_id = result.result_rows[0][0] if result.result_rows else 0 plan.id = (max_id or 0) + 1 plan.created_at = datetime.now() plan.updated_at = datetime.now() query = """ INSERT INTO trading_plans VALUES ( %(id)s, %(plan_name)s, %(status)s, %(created_at)s, %(updated_at)s, %(timeframe)s, %(market_focus)s, %(entry_criteria)s, %(exit_criteria)s, %(stop_loss)s, %(profit_target)s, %(risk_reward_ratio)s, %(trade_frequency)s, %(market_conditions)s, %(indicators_used)s, %(entry_confirmation)s, %(position_sizing)s, %(maximum_drawdown)s, %(max_trades_per_day)s, %(max_trades_per_week)s, %(total_risk_per_trade)s, %(max_portfolio_risk)s, %(adjustments_for_drawdown)s, %(risk_controls)s, %(win_rate)s, %(average_return_per_trade)s, %(profit_factor)s, %(historical_backtest_results)s, %(real_trade_performance)s, %(improvements_needed)s, %(strategy_version)s, %(plan_author)s, %(trade_review_notes)s, %(future_testing_ideas)s, %(sector_focus)s, %(fundamental_criteria)s, %(options_strategy_details)s ) """ params = { 'id': plan.id, 'plan_name': plan.plan_name, 'status': plan.status.value, 'created_at': plan.created_at, 'updated_at': plan.updated_at, 'timeframe': plan.timeframe.value, 'market_focus': plan.market_focus.value, 'entry_criteria': plan.entry_criteria, 'exit_criteria': plan.exit_criteria, 'stop_loss': plan.stop_loss, 'profit_target': plan.profit_target, 'risk_reward_ratio': plan.risk_reward_ratio, 'trade_frequency': plan.trade_frequency.value, 'market_conditions': plan.market_conditions, 'indicators_used': plan.indicators_used, 'entry_confirmation': plan.entry_confirmation, 'position_sizing': plan.position_sizing, 'maximum_drawdown': plan.maximum_drawdown, 'max_trades_per_day': plan.max_trades_per_day, 'max_trades_per_week': plan.max_trades_per_week, 'total_risk_per_trade': plan.total_risk_per_trade, 'max_portfolio_risk': plan.max_portfolio_risk, 'adjustments_for_drawdown': plan.adjustments_for_drawdown, 'risk_controls': plan.risk_controls, 'win_rate': plan.win_rate, 'average_return_per_trade': plan.average_return_per_trade, 'profit_factor': plan.profit_factor, 'historical_backtest_results': plan.historical_backtest_results, 'real_trade_performance': plan.real_trade_performance, 'improvements_needed': plan.improvements_needed, 'strategy_version': plan.strategy_version, 'plan_author': plan.plan_author, 'trade_review_notes': plan.trade_review_notes, 'future_testing_ideas': plan.future_testing_ideas, 'sector_focus': plan.sector_focus, 'fundamental_criteria': plan.fundamental_criteria, 'options_strategy_details': plan.options_strategy_details } client.command(query, params) return plan.id def get_trading_plan(plan_id: int) -> Optional[TradingPlan]: """Get a trading plan by ID""" with create_client() as client: result = client.query(""" SELECT * FROM trading_plans WHERE id = %(id)s """, {'id': plan_id}) rows = result.result_rows if not rows: return None plan = rows[0] return TradingPlan( id=plan[0], plan_name=plan[1], status=PlanStatus(plan[2]), created_at=plan[3], updated_at=plan[4], timeframe=Timeframe(plan[5]), market_focus=MarketFocus(plan[6]), entry_criteria=plan[7], exit_criteria=plan[8], stop_loss=plan[9], profit_target=plan[10], risk_reward_ratio=plan[11], trade_frequency=TradeFrequency(plan[12]), market_conditions=plan[13], indicators_used=plan[14], entry_confirmation=plan[15], position_sizing=plan[16], maximum_drawdown=plan[17], max_trades_per_day=plan[18], max_trades_per_week=plan[19], total_risk_per_trade=plan[20], max_portfolio_risk=plan[21], adjustments_for_drawdown=plan[22], risk_controls=plan[23], win_rate=plan[24], average_return_per_trade=plan[25], profit_factor=plan[26], historical_backtest_results=plan[27], real_trade_performance=plan[28], improvements_needed=plan[29], strategy_version=plan[30], plan_author=plan[31], trade_review_notes=plan[32], future_testing_ideas=plan[33], sector_focus=plan[34], fundamental_criteria=plan[35], options_strategy_details=plan[36] ) def get_all_trading_plans(status: Optional[PlanStatus] = None) -> List[TradingPlan]: """Get all trading plans, optionally filtered by status""" with create_client() as client: try: query = "SELECT * FROM trading_plans" params = {} if status: query += " WHERE status = %(status)s" params['status'] = status.value query += " ORDER BY updated_at DESC" results = client.query(query, params) rows = results.result_rows return [TradingPlan( id=row[0], plan_name=row[1], status=PlanStatus(row[2]), created_at=row[3], updated_at=row[4], timeframe=Timeframe(row[5]), market_focus=MarketFocus(row[6]), entry_criteria=row[7], exit_criteria=row[8], stop_loss=row[9], profit_target=row[10], risk_reward_ratio=row[11], trade_frequency=TradeFrequency(row[12]), market_conditions=row[13], indicators_used=row[14], entry_confirmation=row[15], position_sizing=row[16], maximum_drawdown=row[17], max_trades_per_day=row[18], max_trades_per_week=row[19], total_risk_per_trade=row[20], max_portfolio_risk=row[21], adjustments_for_drawdown=row[22], risk_controls=row[23], win_rate=row[24], average_return_per_trade=row[25], profit_factor=row[26], historical_backtest_results=row[27], real_trade_performance=row[28], improvements_needed=row[29], strategy_version=row[30], plan_author=row[31], trade_review_notes=row[32], future_testing_ideas=row[33], sector_focus=row[34], fundamental_criteria=row[35], options_strategy_details=row[36] ) for row in rows] except Exception as e: print(f"Error retrieving trading plans: {e}") return [] def unlink_trades_from_plan(plan_id: int) -> bool: """Unlink all trades from a trading plan""" with create_client() as client: try: # First update the plan's metrics to NULL plan_update_query = """ ALTER TABLE trading_plans UPDATE win_rate = NULL, average_return_per_trade = NULL, profit_factor = NULL WHERE id = %(plan_id)s """ client.command(plan_update_query, {'plan_id': plan_id}) # Then unlink the trades trades_update_query = """ ALTER TABLE stock_db.trades UPDATE plan_id = NULL WHERE plan_id = %(plan_id)s """ client.command(trades_update_query, {'plan_id': plan_id}) return True except Exception as e: print(f"Error unlinking trades from plan: {e}") return False def delete_trading_plan(plan_id: int) -> bool: """Delete a trading plan by ID""" with create_client() as client: try: # First unlink all trades unlink_trades_from_plan(plan_id) # Then delete the plan query = "ALTER TABLE trading_plans DELETE WHERE id = %(id)s" client.command(query, {'id': plan_id}) return True except Exception as e: print(f"Error deleting trading plan: {e}") return False def link_trades_to_plan(plan_id: int, trade_ids: List[int]) -> bool: """Link existing trades to a trading plan""" with create_client() as client: try: # Format the trade_ids properly for the IN clause trade_ids_str = ", ".join(str(id) for id in trade_ids) query = f""" ALTER TABLE stock_db.trades UPDATE plan_id = {plan_id} WHERE id IN ({trade_ids_str}) """ client.command(query) return True except Exception as e: print(f"Error linking trades to plan: {e}") return False def get_plan_trades(plan_id: int) -> List[dict]: """Get all trades associated with a trading plan""" with create_client() as client: try: # First check if plan_id column exists check_query = """ SELECT name FROM system.columns WHERE database = 'stock_db' AND table = 'trades' AND name = 'plan_id' """ result = client.query(check_query) if not result.result_rows: # Add plan_id column if it doesn't exist alter_query = """ ALTER TABLE stock_db.trades ADD COLUMN IF NOT EXISTS plan_id Nullable(UInt32) """ client.command(alter_query) print("Added plan_id column to trades table") # Now query the trades query = """ SELECT * FROM stock_db.trades WHERE plan_id = %(plan_id)s ORDER BY entry_date DESC """ result = client.query(query, {'plan_id': plan_id}) return [dict(zip( ['id', 'position_id', 'ticker', 'entry_date', 'shares', 'entry_price', 'target_price', 'stop_loss', 'strategy', 'order_type', 'direction', 'followed_rules', 'entry_reason', 'exit_price', 'exit_date', 'exit_reason', 'notes', 'created_at', 'plan_id'], row )) for row in result.result_rows] except Exception as e: print(f"Error in get_plan_trades: {e}") return [] def calculate_plan_metrics(plan_id: int) -> dict: """Calculate performance metrics for a trading plan""" trades = get_plan_trades(plan_id) if not trades: return {} total_trades = len(trades) winning_trades = sum(1 for t in trades if t['exit_price'] and (t['exit_price'] - t['entry_price']) * t['shares'] > 0) total_profit = sum((t['exit_price'] - t['entry_price']) * t['shares'] for t in trades if t['exit_price']) gross_profits = sum((t['exit_price'] - t['entry_price']) * t['shares'] for t in trades if t['exit_price'] and (t['exit_price'] - t['entry_price']) * t['shares'] > 0) gross_losses = abs(sum((t['exit_price'] - t['entry_price']) * t['shares'] for t in trades if t['exit_price'] and (t['exit_price'] - t['entry_price']) * t['shares'] < 0)) return { 'total_trades': total_trades, 'winning_trades': winning_trades, 'win_rate': (winning_trades / total_trades * 100) if total_trades > 0 else 0, 'total_profit': total_profit, 'average_return': total_profit / total_trades if total_trades > 0 else 0, 'profit_factor': gross_profits / gross_losses if gross_losses > 0 else float('inf') } def update_trading_plan(plan: TradingPlan) -> bool: """Update an existing trading plan""" if not plan.id: raise ValueError("Cannot update plan without ID") with create_client() as client: plan.updated_at = datetime.now() query = """ ALTER TABLE trading_plans UPDATE plan_name = %(plan_name)s, status = %(status)s, updated_at = %(updated_at)s, timeframe = %(timeframe)s, market_focus = %(market_focus)s, entry_criteria = %(entry_criteria)s, exit_criteria = %(exit_criteria)s, stop_loss = %(stop_loss)s, profit_target = %(profit_target)s, risk_reward_ratio = %(risk_reward_ratio)s, trade_frequency = %(trade_frequency)s, market_conditions = %(market_conditions)s, indicators_used = %(indicators_used)s, entry_confirmation = %(entry_confirmation)s, position_sizing = %(position_sizing)s, maximum_drawdown = %(maximum_drawdown)s, max_trades_per_day = %(max_trades_per_day)s, max_trades_per_week = %(max_trades_per_week)s, total_risk_per_trade = %(total_risk_per_trade)s, max_portfolio_risk = %(max_portfolio_risk)s, adjustments_for_drawdown = %(adjustments_for_drawdown)s, risk_controls = %(risk_controls)s, win_rate = %(win_rate)s, average_return_per_trade = %(average_return_per_trade)s, profit_factor = %(profit_factor)s, historical_backtest_results = %(historical_backtest_results)s, real_trade_performance = %(real_trade_performance)s, improvements_needed = %(improvements_needed)s, strategy_version = %(strategy_version)s, plan_author = %(plan_author)s, trade_review_notes = %(trade_review_notes)s, future_testing_ideas = %(future_testing_ideas)s, sector_focus = %(sector_focus)s, fundamental_criteria = %(fundamental_criteria)s, options_strategy_details = %(options_strategy_details)s WHERE id = %(id)s """ params = { 'id': plan.id, 'plan_name': plan.plan_name, 'status': plan.status.value, 'updated_at': plan.updated_at, 'timeframe': plan.timeframe.value, 'market_focus': plan.market_focus.value, 'entry_criteria': plan.entry_criteria, 'exit_criteria': plan.exit_criteria, 'stop_loss': plan.stop_loss, 'profit_target': plan.profit_target, 'risk_reward_ratio': plan.risk_reward_ratio, 'trade_frequency': plan.trade_frequency.value, 'market_conditions': plan.market_conditions, 'indicators_used': plan.indicators_used, 'entry_confirmation': plan.entry_confirmation, 'position_sizing': plan.position_sizing, 'maximum_drawdown': plan.maximum_drawdown, 'max_trades_per_day': plan.max_trades_per_day, 'max_trades_per_week': plan.max_trades_per_week, 'total_risk_per_trade': plan.total_risk_per_trade, 'max_portfolio_risk': plan.max_portfolio_risk, 'adjustments_for_drawdown': plan.adjustments_for_drawdown, 'risk_controls': plan.risk_controls, 'win_rate': plan.win_rate, 'average_return_per_trade': plan.average_return_per_trade, 'profit_factor': plan.profit_factor, 'historical_backtest_results': plan.historical_backtest_results, 'real_trade_performance': plan.real_trade_performance, 'improvements_needed': plan.improvements_needed, 'strategy_version': plan.strategy_version, 'plan_author': plan.plan_author, 'trade_review_notes': plan.trade_review_notes, 'future_testing_ideas': plan.future_testing_ideas, 'sector_focus': plan.sector_focus, 'fundamental_criteria': plan.fundamental_criteria, 'options_strategy_details': plan.options_strategy_details } client.command(query, params) return True