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 argparse import RawDescriptionHelpFormatter
from datetime import date
from datetime import timedelta
from typing import Any
from typing import Dict

from common.logger import setup_logging


def parse_args():
    parser = ArgumentParser(
        description=__doc__, formatter_class=RawDescriptionHelpFormatter
    )
    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)