Bobby/stock_system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Consolidate initialize_scanner function and remove duplicate

Browse Source
This commit is contained in:
Bobby (aider) 2025-02-12 19:46:35 -08:00
parent ec50fc1ebc
commit 7d7262fd8c
1 changed files with 1 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
src/utils/data_utils.py
View File

@ -5,6 +5,7 @@ from datetime import datetime, timedelta
from db.db_connection import create_client from db.db_connection import create_client
from screener.user_input import get_interval_choice, get_date_range from screener.user_input import get_interval_choice, get_date_range
from trading.position_calculator import PositionCalculator from trading.position_calculator import PositionCalculator
from utils.scanner_utils import initialize_scanner
from typing import Optional from typing import Optional
@ -203,42 +204,6 @@ def save_signals_to_csv(signals: list, scanner_name: str) -> None:
df_signals.to_csv(output_file, index=False) df_signals.to_csv(output_file, index=False)
print(f"\nSaved {len(signals)} signals to {output_file}") print(f"\nSaved {len(signals)} signals to {output_file}")
def initialize_scanner(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> tuple:
"""
Initialize common scanner components
Args:
min_price (float): Minimum stock price
max_price (float): Maximum stock price
min_volume (int): Minimum trading volume
portfolio_size (float, optional): Portfolio size for position sizing
Returns:
tuple: (interval, start_date, end_date, qualified_stocks, calculator)
"""
print(f"\nScanning for stocks ${min_price:.2f}-${max_price:.2f} with min volume {min_volume:,}")
interval = get_interval_choice()
start_date, end_date = get_date_range()
qualified_stocks = get_qualified_stocks(start_date, end_date, min_price, max_price, min_volume)
if not qualified_stocks:
print("No stocks found matching criteria.")
return None, None, None, None, None
print(f"\nFound {len(qualified_stocks)} stocks matching criteria")
# Initialize position calculator if portfolio size provided
calculator = None
if portfolio_size and portfolio_size > 0:
calculator = PositionCalculator(
account_size=portfolio_size,
risk_percentage=1.0,
stop_loss_percentage=7.0
)
return interval, start_date, end_date, qualified_stocks, calculator
def process_signal_data(ticker: str, signal_data: dict, current_volume: int, def process_signal_data(ticker: str, signal_data: dict, current_volume: int,
last_update: int, stock_type: str, calculator: PositionCalculator = None) -> dict: last_update: int, stock_type: str, calculator: PositionCalculator = None) -> dict: