ib_es_short_straddle_bot.py

#!/usr/bin/env python3
"""
Short straddles with adjustments bot running on IB

Usage:
./ib_es_short_straddle_bot.py -h

./ib_es_short_straddle_bot.py --dte 3 -v # To log INFO messages
./ib_es_short_straddle_bot.py --dte 3 -vv # To log DEBUG messages
"""

import logging
import time
from argparse import ArgumentParser
from datetime import date, timedelta
from typing import Any, Dict

from common import RawTextWithDefaultsFormatter
from common.logger import setup_logging


def parse_args():
    parser = ArgumentParser(
        description=__doc__, formatter_class=RawTextWithDefaultsFormatter
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="count",
        default=0,
        dest="verbose",
        help="Increase verbosity of logging output",
    )
    parser.add_argument(
        "--dte",
        type=int,
        required=True,
        help="Days to Expiry",
    )
    return parser.parse_args()


def determine_trading_day(dte: int) -> str:
    from_date = date.today()
    target_date = from_date + timedelta(days=dte)

    while target_date.weekday() > 4:  # 5 = Saturday, 6 = Sunday
        target_date += timedelta(days=1)

    return target_date


def analyze_option_chain(trading_day: str) -> Dict[str, Any]:
    # Analyze option chain and return relevant data
    pass


def open_position(option_data: Dict[str, Any]) -> Dict[str, Any]:
    # Open the short straddle position
    # Return position details
    pass


def calculate_break_even_points(position: Dict[str, Any]) -> Dict[str, float]:
    # Calculate and return break-even points
    pass


def monitor_position(position: Dict[str, Any]) -> Dict[str, Any]:
    # Monitor current prices and calculate position value
    pass


def check_adjustment_needed(
    monitoring_data: Dict[str, Any], initial_difference: float
) -> bool:
    # Check if adjustment is needed
    pass


def adjust_position(
    position: Dict[str, Any], monitoring_data: Dict[str, Any]
) -> Dict[str, Any]:
    # Perform position adjustment
    # Return updated position
    pass


def check_exit_conditions(
    monitoring_data: Dict[str, Any], initial_credit: float
) -> bool:
    # Check if exit conditions are met
    pass


def close_position(position: Dict[str, Any]) -> Dict[str, Any]:
    # Close the position and generate summary
    pass


def run_bot(dte: int):
    trading_day = determine_trading_day(dte)
    option_data = analyze_option_chain(trading_day)
    position = open_position(option_data)
    initial_credit = position["initial_credit"]
    initial_difference = position["initial_difference"]

    while True:
        monitoring_data = monitor_position(position)

        if check_adjustment_needed(monitoring_data, initial_difference):
            position = adjust_position(position, monitoring_data)

        if check_exit_conditions(monitoring_data, initial_credit):
            summary = close_position(position)
            print(summary)
            break

        time.sleep(300)  # Wait for 5 minutes before next check


def main(args):
    logging.debug(f"Running bot with verbosity level: {args.verbose}")
    run_bot(args.dte)


if __name__ == "__main__":
    args = parse_args()
    setup_logging(args.verbose)
    main(args)