Based on patterns observed in this Binance Options Trading Bot project, here's a comprehensive Python reference covering essential concepts and advanced patterns.
- Basic Data Types
- Lists
- Dictionaries
- Classes and Objects
- Type Hints
- Functions
- Exception Handling
- List Comprehensions
- Lambda Functions
- Decorators
- Context Managers
- Threading
- String Formatting
- File Operations
- Import Statements
# Integers and Floats
strike_price = 50000 # int
option_price = 1250.75 # float
# Strings
symbol = "BTCUSDT-240624-50000-C"
expiry_code = "240624"
# Booleans
is_call = True
is_in_the_money = False
# None type
no_data = NoneLists are ordered, mutable collections used extensively in this project for managing option symbols and data.
# Creating lists
option_symbols = [] # Empty list
strikes = [30000, 35000, 40000, 45000, 50000]
option_types = ["C", "P"]
# List operations
symbols = ["BTCUSDT-240624-30000-C", "BTCUSDT-240624-35000-C"]
symbols.append("BTCUSDT-240624-40000-C") # Add item
first_symbol = symbols[0] # Access by index
symbols.remove("BTCUSDT-240624-30000-C") # Remove item
# List slicing
first_two = symbols[:2] # First 2 items
last_two = symbols[-2:] # Last 2 items
# List comprehensions (covered in detail later)
squares = [x**2 for x in range(10)]
even_numbers = [x for x in range(20) if x % 2 == 0]
# Iterating through lists
for symbol in symbols:
print(f"Processing: {symbol}")
# List methods commonly used in the project
all_symbols = ["BTCUSDT-240624-30000-C", "BTCUSDT-240624-35000-C"]
all_symbols.clear() # Remove all items
copied_symbols = all_symbols.copy() # Shallow copy
length = len(all_symbols) # Get lengthDictionaries are heavily used in this project for organizing option data, prices, and configuration.
# Creating dictionaries
option_data = {} # Empty dict
prices = {"bid": 1000.50, "ask": 1001.25}
option_info = {
"symbol": "BTCUSDT-240624-50000-C",
"strike": 50000,
"type": "C",
"expiry": "240624"
}
# Dictionary operations
option_data["bid_price"] = 1000.50 # Add/update value
current_bid = option_data.get("bid_price", 0) # Get with default
current_ask = option_data["ask_price"] # Direct access
# Dictionary methods
keys = prices.keys() # Get all keys
values = prices.values() # Get all values
items = prices.items() # Get key-value pairs
# Iterating through dictionaries
for key, value in prices.items():
print(f"{key}: {value}")
# Dictionary comprehensions
squares_dict = {x: x**2 for x in range(5)}
filtered_options = {k: v for k, v in option_info.items() if v is not None}
# Nested dictionaries (used in StreamManager)
option_prices = {
"btcusdt-240624-50000-c": {"bid": 1000.50, "ask": 1001.25},
"btcusdt-240624-55000-c": {"bid": 800.75, "ask": 801.50}
}
# defaultdict (used in the project)
from collections import defaultdict
OPTION_SYMBOLS = defaultdict(list) # Default value is empty list
OPTION_SYMBOLS["240624"].append("BTCUSDT-240624-50000-C")The project demonstrates both regular classes and dataclasses for modeling financial data.
# Regular class (OptionSymbol example)
class OptionSymbol:
def __init__(self, symbol: str, expiry: str, strike: int, option_type: str):
self.symbol = symbol
self.expiry = expiry
self.strike = strike
self.option_type = option_type
self.bid_price = 0
self.ask_price = 0
def update_prices(self) -> None:
# Method to update option prices
pass
def get_mid_price(self) -> float:
if self.bid_price and self.ask_price:
return (self.bid_price + self.ask_price) / 2
return 0
# Using the class
option = OptionSymbol("BTCUSDT-240624-50000-C", "240624", 50000, "C")
option.bid_price = 1000.50
option.ask_price = 1001.25
mid_price = option.get_mid_price()
# Dataclass (ExpiryInfo example)
from dataclasses import dataclass
@dataclass(frozen=True)
class ExpiryInfo:
"""Container for expiry metadata."""
code: str # Format YYMMDD
display: str # Human readable label
# Using dataclass
expiry = ExpiryInfo("240624", "June 24, 2024")
print(f"Expiry code: {expiry.code}")
print(f"Display: {expiry.display}")
# Class inheritance
class BaseOption:
def __init__(self, symbol: str):
self.symbol = symbol
class CallOption(BaseOption):
def __init__(self, symbol: str, strike: int):
super().__init__(symbol)
self.strike = strike
self.option_type = "C"
# Static methods and class methods
class OptionUtils:
@staticmethod
def format_price(price: float) -> str:
return f"${price",.2f"}"
@classmethod
def from_string(cls, symbol_str: str):
# Parse symbol string and create instance
return cls(symbol_str)The project extensively uses type hints for better code clarity and IDE support.
from typing import Dict, List, Optional, Tuple, Any, Callable
from datetime import datetime
# Basic type hints
def get_option_price(symbol: str) -> float:
pass
def get_multiple_prices(symbols: List[str]) -> Dict[str, float]:
pass
# Optional types
def find_option(symbol: str) -> Optional[OptionSymbol]:
pass
# Union types
from typing import Union
def process_price(price: Union[float, None]) -> str:
pass
# Generic types
from typing import TypeVar
T = TypeVar('T')
def first_item(items: List[T]) -> T:
return items[0]
# Callable types (for function parameters)
def subscribe_to_updates(callback: Callable[[dict], None]) -> None:
pass
# Complex type hints used in the project
def build_option_subscription(
symbols: Iterable[str]
) -> Tuple[SpotWebsocketStreamClient, List[str]]:
passVarious function patterns used throughout the project.
# Basic function
def calculate_mid_price(bid: float, ask: float) -> float:
return (bid + ask) / 2
# Function with default parameters
def get_option_chain(expiry: str = "240624", max_strikes: int = 20) -> List[OptionSymbol]:
pass
# Variable arguments
def combine_symbols(*symbols: str) -> str:
return "-".join(symbols)
# Keyword arguments
def create_option(symbol: str, **kwargs) -> OptionSymbol:
return OptionSymbol(symbol=symbol, **kwargs)
# Generator functions
def generate_strikes(start: int, end: int, step: int = 1000):
current = start
while current <= end:
yield current
current += step
# Using generators
for strike in generate_strikes(30000, 50000):
print(f"Strike: {strike}")
# Lambda functions (covered in detail later)
price_filter = lambda x: x > 1000
filtered_prices = list(filter(price_filter, [800, 1200, 1500]))Proper error handling patterns used in the project.
# Basic exception handling
try:
price = float("invalid_price")
except ValueError as e:
print(f"Invalid price format: {e}")
price = 0
# Multiple exception types
try:
# Some operation that might fail
result = risky_operation()
except (ValueError, TypeError) as e:
print(f"Data error: {e}")
except ConnectionError as e:
print(f"Network error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
# Finally block (always executes)
try:
stream_manager = StreamManager()
stream_manager.start()
except Exception as e:
print(f"Failed to start stream: {e}")
finally:
# Cleanup code
if 'stream_manager' in locals():
stream_manager.stop()
# Raising custom exceptions
class InvalidOptionSymbol(Exception):
pass
def validate_symbol(symbol: str) -> None:
if not symbol or "-" not in symbol:
raise InvalidOptionSymbol(f"Invalid symbol format: {symbol}")
# Context managers for exception handling
from contextlib import contextmanager
@contextmanager
def managed_stream():
stream = None
try:
stream = StreamManager()
yield stream
finally:
if stream:
stream.stop()Powerful and concise way to create lists, used throughout the project.
# Basic list comprehension
numbers = [1, 2, 3, 4, 5]
squares = [x**2 for x in numbers] # [1, 4, 9, 16, 25]
# With conditions
even_squares = [x**2 for x in numbers if x % 2 == 0] # [4, 16]
# Nested comprehensions
matrix = [[1, 2], [3, 4]]
flattened = [x for row in matrix for x in row] # [1, 2, 3, 4]
# Dictionary comprehensions
names = ["Alice", "Bob", "Charlie"]
name_lengths = {name: len(name) for name in names}
# Set comprehensions
unique_lengths = {len(name) for name in names}
# Complex example from the project
option_symbols = ["BTCUSDT-240624-30000-C", "BTCUSDT-240624-35000-C"]
strikes = [int(symbol.split("-")[2]) for symbol in option_symbols]
# Filtering with comprehensions
active_options = [
option for option in all_options
if option.bid_price > 0 and option.ask_price > 0
]Anonymous functions used for simple operations.
# Basic lambda
square = lambda x: x**2
result = square(5) # 25
# Lambda with multiple parameters
add = lambda x, y: x + y
result = add(3, 4) # 7
# Using with built-in functions
numbers = [1, 4, 2, 8, 5, 7]
sorted_numbers = sorted(numbers) # [1, 2, 4, 5, 7, 8]
# Sort by custom key
words = ["apple", "Banana", "cherry"]
sorted_words = sorted(words, key=lambda x: x.lower())
# Filter with lambda
even_numbers = list(filter(lambda x: x % 2 == 0, numbers))
# Map with lambda
squares = list(map(lambda x: x**2, numbers))
# Lambda in defaultdict (used in project)
from collections import defaultdict
OPTION_SYMBOLS = defaultdict(lambda: []) # Default empty listDecorators modify function behavior, used in the project for UI refresh and startup hooks.
# Basic decorator
def timer(func):
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
end = time.time()
print(f"{func.__name__} took {end - start:.2f} seconds")
return result
return wrapper
@timer
def slow_function():
time.sleep(1)
return "Done"
# Class method decorators
class OptionManager:
@staticmethod
def format_price(price: float) -> str:
return f"${price",.2f"}"
@classmethod
def create_from_symbol(cls, symbol: str):
return cls(symbol)
# Property decorators
class Option:
def __init__(self, symbol: str):
self._symbol = symbol
@property
def symbol(self) -> str:
return self._symbol
@symbol.setter
def symbol(self, value: str) -> None:
if not value:
raise ValueError("Symbol cannot be empty")
self._symbol = value
# Dataclass decorator (used in project)
from dataclasses import dataclass
@dataclass(frozen=True)
class ExpiryInfo:
code: str
display: strUsed for resource management, especially important for streams and connections.
# Using with statement
with open("data.txt", "r") as file:
data = file.read()
# Custom context manager
from contextlib import contextmanager
@contextmanager
def stream_manager():
manager = StreamManager()
try:
yield manager
finally:
manager.stop()
# Using custom context manager
with stream_manager() as manager:
prices = manager.get_prices()
# Context manager class
class DatabaseConnection:
def __enter__(self):
self.connection = create_connection()
return self.connection
def __exit__(self, exc_type, exc_val, exc_tb):
self.connection.close()
return False # Don't suppress exceptions
# Using context manager class
with DatabaseConnection() as conn:
data = conn.query("SELECT * FROM options")Used in the project for managing websocket streams safely.
import threading
from threading import Lock
# Basic threading
def worker():
print("Worker thread running")
thread = threading.Thread(target=worker)
thread.start()
thread.join() # Wait for completion
# Locks for thread safety (used in StreamManager)
class SafeCounter:
def __init__(self):
self._value = 0
self._lock = Lock()
def increment(self):
with self._lock:
self._value += 1
def get_value(self):
with self._lock:
return self._value
# Thread-safe data structures
prices_lock = Lock()
option_prices = {}
def update_price(symbol: str, price: float):
with prices_lock:
option_prices[symbol] = price
def get_price(symbol: str) -> float:
with prices_lock:
return option_prices.get(symbol, 0)Various string formatting techniques used in the project.
# Old-style formatting
price = 1000.50
formatted = "Price: %0.2f" % price
# .format() method
formatted = "Price: {:.2f}".format(price)
formatted = "Symbol: {}, Price: {:.2f}".format("BTCUSDT", price)
# f-strings (most common in modern Python)
symbol = "BTCUSDT-240624-50000-C"
formatted = f"Symbol: {symbol}, Price: {price:.2f}"
formatted = f"Option {symbol} expires in {days_left} days"
# String methods
symbol = "BTCUSDT-240624-50000-C"
expiry = symbol.split("-")[1] # "240624"
symbol_type = symbol[-1] # "C"
# Join strings
parts = ["BTCUSDT", "240624", "50000", "C"]
symbol = "-".join(parts)
# String formatting with conditions
status = "ITM" if strike < spot_price else "OTM"
formatted = f"Option is {status}"File handling patterns for configuration and data storage.
# Reading files
with open("config.txt", "r") as file:
content = file.read()
with open("data.txt", "r") as file:
lines = file.readlines() # Read all lines
# Writing files
with open("output.txt", "w") as file:
file.write("Option data\n")
# Appending to files
with open("log.txt", "a") as file:
file.write(f"{datetime.now()}: Price update\n")
# JSON handling (common for API data)
import json
# Read JSON
with open("config.json", "r") as file:
config = json.load(file)
# Write JSON
data = {"symbol": "BTCUSDT", "price": 50000}
with open("data.json", "w") as file:
json.dump(data, file, indent=2)
# CSV handling
import csv
# Read CSV
with open("options.csv", "r") as file:
reader = csv.DictReader(file)
for row in reader:
print(row["symbol"])
# Write CSV
with open("output.csv", "w", newline="") as file:
writer = csv.writer(file)
writer.writerow(["Symbol", "Price", "Strike"])
writer.writerow(["BTCUSDT-240624-50000-C", "1000.50", "50000"])Various import patterns used in the project.
# Standard library imports
import time
import json
from datetime import datetime
# Third-party imports
from nicegui import ui, app
from binance.websocket.spot.websocket_stream import SpotWebsocketStreamClient
# Local imports
from option_chain.models import OptionSymbol, ExpiryInfo
from option_chain.stream import StreamManager
# Relative imports
from .models import OptionSymbol
from ..utils import helper_function
# Conditional imports
try:
import optional_library
except ImportError:
optional_library = None
# Import with alias
import pandas as pd
import numpy as np
# From import with multiple items
from typing import Dict, List, Optional, Tuple, Any
# Future imports (used in project)
from __future__ import annotationsclass StreamManager:
def __init__(self):
self._option_prices: Dict[str, Dict[str, Optional[float]]] = {}
self._option_lock = Lock()
self._option_symbols: List[str] = []
def get_option_prices_immediate(self, symbol: str) -> Tuple[Optional[float], Optional[float]]:
symbol_key = _option_symbol_key(symbol)
with self._option_lock:
prices = self._option_prices.get(symbol_key)
if not prices:
return None, None
return prices.get("ask"), prices.get("bid")@ui.refreshable
def TimeDisplay():
ui.label(f"Current Time: {time.strftime('%Y-%m-%d %H:%M:%S')}").classes('text-sm text-gray-600')
@ui.refreshable
def OnUIStart():
TimeDisplay()
# UI components heredef _handle_option_message(message: str, handler: Optional[Callable[[dict], None]]) -> None:
try:
payload = json.loads(message)
except json.JSONDecodeError:
logging.debug("Option stream non-JSON message: %s", message)
return
# Process messageThis cheat sheet covers the essential Python concepts and patterns used in this Binance Options Trading Bot project. The project demonstrates professional Python development practices including proper type hints, error handling, threading safety, and clean code organization.