feat: Resolve circular imports and create scanner_utils module

This commit is contained in:
Bobby (aider) 2025-02-10 10:01:46 -08:00
parent 3dbaed0399
commit 529875527b
4 changed files with 65 additions and 18 deletions

View File

@ -1,10 +1 @@
# Add explicit imports for scanner modules
from .t_atr_ema import run_atr_ema_scanner
from .t_atr_ema_v2 import run_atr_ema_scanner_v2
from .t_sunnyband import run_sunny_scanner
__all__ = [
'run_atr_ema_scanner',
'run_atr_ema_scanner_v2',
'run_sunny_scanner'
]
# Empty file

View File

@ -1,9 +1,6 @@
import pandas as pd
from utils.data_utils import (
get_stock_data, validate_signal_date, print_signal,
save_signals_to_csv, get_qualified_stocks,
initialize_scanner, process_signal_data
)
from utils.scanner_utils import initialize_scanner, process_signal_data
from utils.data_utils import get_stock_data, validate_signal_date, print_signal, save_signals_to_csv
from indicators.three_atr_ema import ThreeATREMAIndicator
def check_entry_signal(df: pd.DataFrame) -> list:

View File

@ -1,3 +1 @@
from .data_utils import get_stock_data
__all__ = ['get_stock_data']
# Empty file

View File

@ -0,0 +1,61 @@
from datetime import datetime, timedelta
from utils.data_utils import get_user_input, get_stock_data, get_qualified_stocks
from screener.user_input import get_interval_choice, get_date_range
from trading.position_calculator import PositionCalculator
def initialize_scanner(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> tuple:
"""
Initialize common scanner components
"""
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,
last_update: int, stock_type: str, calculator: PositionCalculator = None) -> dict:
"""
Process and format signal data consistently
"""
entry_data = {
'ticker': ticker,
'entry_price': signal_data['price'],
'target_price': signal_data.get('ema', signal_data.get('upper_band')), # Handle both ATR and Sunny
'volume': current_volume,
'signal_date': signal_data.get('date', datetime.now()),
'stock_type': stock_type,
'last_update': datetime.fromtimestamp(last_update/1000000000)
}
if calculator:
position = calculator.calculate_position_size(entry_data['entry_price'])
potential_profit = (entry_data['target_price'] - entry_data['entry_price']) * position['shares']
entry_data.update({
'shares': position['shares'],
'position_size': position['position_value'],
'stop_loss': position['stop_loss'],
'risk_amount': position['potential_loss'],
'profit_amount': potential_profit,
'risk_reward_ratio': abs(potential_profit / position['potential_loss']) if position['potential_loss'] != 0 else 0
})
return entry_data