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

refactor: Modularize scanner initialization and signal processing

Browse Source
This commit is contained in:
Bobby (aider) 2025-02-09 12:16:27 -08:00
parent d8b41c8d8a
commit 7f105bfd54
2 changed files with 87 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
src/screener/t_atr_ema_v2.py
View File

@ -70,32 +70,15 @@ def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int,
min_volume (int): Minimum trading volume
portfolio_size (float, optional): Portfolio size for position sizing
"""
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()
start_ts = int(start_date.timestamp() * 1000000000)
end_ts = int(end_date.timestamp() * 1000000000)
try:
qualified_stocks = get_qualified_stocks(start_date, end_date, min_price, max_price, min_volume)
# Initialize scanner components
interval, start_date, end_date, qualified_stocks, calculator = initialize_scanner(
min_price, max_price, min_volume, portfolio_size
)
if not qualified_stocks:
print("No stocks found matching criteria.")
return
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
)
entry_signals = []
for ticker, current_price, current_volume, last_update, stock_type in qualified_stocks:
@ -107,28 +90,11 @@ def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int,
signals = check_entry_signal(df)
for signal, signal_date, signal_data in signals:
entry_data = {
'ticker': ticker,
'entry_price': signal_data['price'],
'target_price': signal_data['ema'],
'volume': current_volume,
'signal_date': signal_date,
'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
})
signal_data['date'] = signal_date
entry_data = process_signal_data(
ticker, signal_data, current_volume,
last_update, stock_type, calculator
)
entry_signals.append(entry_data)
print_signal(entry_data)

77
src/utils/data_utils.py
View File

@ -134,6 +134,83 @@ def save_signals_to_csv(signals: list, scanner_name: str) -> None:
df_signals.to_csv(output_file, index=False)
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,
last_update: int, stock_type: str, calculator: PositionCalculator = None) -> dict:
"""
Process and format signal data consistently
Args:
ticker (str): Stock ticker
signal_data (dict): Raw signal data
current_volume (int): Current trading volume
last_update (int): Last update timestamp
stock_type (str): Stock type/label
calculator (PositionCalculator, optional): Position calculator instance
Returns:
dict: Processed signal data
"""
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
def get_stock_data(ticker: str, start_date: datetime, end_date: datetime, interval: str) -> pd.DataFrame:
"""
Fetch and resample stock data based on the chosen interval