Trader/test_value_averaging.py at main · Bonney96/Trader · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
from connection import AlpacaConnection
from strategies.value_averaging import ValueAveragingStrategy
import logging
from config import LOG_LEVEL, LOG_FILE
import time
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=getattr(logging, LOG_LEVEL),
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(LOG_FILE),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

def get_market_status(api):
    """Get and display market status"""
    try:
        clock = api.get_clock()
        next_open = clock.next_open
        next_close = clock.next_close

        if clock.is_open:
            logger.info(f"✅ Market is OPEN | Closes at: {next_close}")
        else:
            logger.info(f"❌ Market is CLOSED | Next open: {next_open}")

        return clock.is_open
    except Exception as e:
        logger.error(f"❌ Error getting market status: {str(e)}")
        return False

def wait_for_market_open(api, check_interval=60):
    """Wait for market to open"""
    while True:
        if get_market_status(api):
            logger.info("🚀 Market is now open!")
            return True
        logger.info(f"⏳ Waiting for market to open... Checking again in {check_interval} seconds")
        time.sleep(check_interval)

def test_strategy():
    """Test the Value Averaging strategy for PLTR"""
    # Initialize connection
    alpaca = AlpacaConnection()

    try:
        # Connect to Alpaca
        logger.info("🔄 Testing Alpaca connection...")
        if not alpaca.connect():
            logger.error("❌ Failed to connect to Alpaca")
            return False

        # Check market status and wait if closed
        is_market_open = get_market_status(alpaca.api)
        if not is_market_open:
            logger.warning("🚦 Market closed. Waiting for the next trading session...")
            wait_for_market_open(alpaca.api)

        # Initialize strategy
        logger.info("📈 Initializing Value Averaging strategy for PLTR...")
        strategy = ValueAveragingStrategy(
            api=alpaca.api,
            symbol='PLTR',
            weekly_target=100,  # Target $100 investment per week
            stop_loss_pct=0.1,  # 10% stop loss
            take_profit_pct=0.2  # 20% take profit
        )

        # Run strategy for 5 minutes
        logger.info("🚀 Running strategy for 5 minutes...")
        start_time = time.time()
        while time.time() - start_time < 300:  # Run for 5 minutes
            try:
                # Check market status
                if not get_market_status(alpaca.api):
                    logger.warning("🚦 Market closed. Waiting for next session...")
                    wait_for_market_open(alpaca.api)

                # Get current price
                current_price = strategy.get_current_price()
                if current_price is None:
                    logger.warning("⚠️ Could not fetch stock price. Retrying...")
                    time.sleep(30)
                    continue

                logger.info(f"💲 Current PLTR Price: ${current_price:.2f}")

                # Calculate investment needed
                investment_needed = strategy.calculate_investment_needed()
                logger.info(f"🔢 Investment Needed: ${investment_needed:.2f}")

                # Execute trade if needed
                if investment_needed > 0:
                    strategy.execute_trade(investment_needed)
                    logger.info(f"✅ Executed trade for ${investment_needed:.2f}")

                # Wait for 1 minute before next iteration
                time.sleep(60)

            except Exception as e:
                logger.error(f"⚠️ Error in strategy loop: {str(e)}")
                time.sleep(60)

        logger.info("🏁 Strategy test completed!")
        return True

    except KeyboardInterrupt:
        logger.info("🛑 Strategy test interrupted by user")
        return False
    except Exception as e:
        logger.error(f"⚠️ Unexpected error: {str(e)}")
        return False
    finally:
        # Clean up
        if alpaca.is_connected():
            alpaca.disconnect()

if __name__ == "__main__":
    test_strategy()