feat: Add SunnyBand scanner and multi-system menu to main.py
This commit is contained in:
parent
fc7a6e7343
commit
a3a87c1e7d
147
src/main.py
147
src/main.py
@ -2,72 +2,99 @@ import datetime
|
||||
from screener.data_fetcher import validate_date_range, fetch_financial_data, get_stocks_in_time_range
|
||||
from screener.c_canslim import check_quarterly_earnings, check_return_on_equity, check_sales_growth
|
||||
from screener.a_canslim import check_annual_eps_growth
|
||||
from screener.l_canslim import check_industry_leadership # ✅ NEW: Import L Score function
|
||||
from screener.i_canslim import check_institutional_sponsorship # ✅ NEW: Import I Score function
|
||||
from screener.l_canslim import check_industry_leadership
|
||||
from screener.i_canslim import check_institutional_sponsorship
|
||||
from screener.csv_appender import append_scores_to_csv
|
||||
from screener.screeners import SCREENERS # Import categories
|
||||
from screener.user_input import get_user_screener_selection # Import function
|
||||
from screener.screeners import SCREENERS
|
||||
from screener.user_input import get_user_screener_selection
|
||||
|
||||
def get_float_input(prompt: str) -> float:
|
||||
while True:
|
||||
try:
|
||||
return float(input(prompt))
|
||||
except ValueError:
|
||||
print("Please enter a valid number")
|
||||
|
||||
def get_scanner_parameters():
|
||||
"""Get user input for scanner parameters"""
|
||||
min_price = get_float_input("Enter minimum stock price ($): ")
|
||||
max_price = get_float_input("Enter maximum stock price ($): ")
|
||||
min_volume = int(input("Enter minimum volume: "))
|
||||
return min_price, max_price, min_volume
|
||||
|
||||
def main():
|
||||
# 1️⃣ Ask user for start and end date
|
||||
user_start_date = input("Enter start date (YYYY-MM-DD): ")
|
||||
user_end_date = input("Enter end date (YYYY-MM-DD): ")
|
||||
print("\nStock Analysis System")
|
||||
print("1. Run CANSLIM Screener")
|
||||
print("2. Run SunnyBand Scanner")
|
||||
print("3. Launch Trading System")
|
||||
print("4. Exit")
|
||||
|
||||
# 2️⃣ Validate and adjust date range if needed
|
||||
start_date, end_date = validate_date_range(user_start_date, user_end_date, required_quarters=4)
|
||||
|
||||
# 3️⃣ Get selected screeners & customization preferences
|
||||
selected_screeners = get_user_screener_selection()
|
||||
print(f"\n✅ Selected Screeners: {selected_screeners}\n") # ✅ DEBUG LOG
|
||||
|
||||
# 4️⃣ Get all stock symbols dynamically
|
||||
symbol_list = get_stocks_in_time_range(start_date, end_date)
|
||||
choice = input("\nSelect an option (1-4): ")
|
||||
|
||||
if not symbol_list:
|
||||
print("No stocks found within the given date range.")
|
||||
if choice == "1":
|
||||
# 1️⃣ Ask user for start and end date
|
||||
user_start_date = input("Enter start date (YYYY-MM-DD): ")
|
||||
user_end_date = input("Enter end date (YYYY-MM-DD): ")
|
||||
|
||||
# 2️⃣ Validate and adjust date range if needed
|
||||
start_date, end_date = validate_date_range(user_start_date, user_end_date, required_quarters=4)
|
||||
selected_screeners = get_user_screener_selection()
|
||||
symbol_list = get_stocks_in_time_range(start_date, end_date)
|
||||
|
||||
if not symbol_list:
|
||||
print("No stocks found within the given date range.")
|
||||
return
|
||||
|
||||
print(f"Processing {len(symbol_list)} stocks within the given date range...\n")
|
||||
|
||||
for symbol in symbol_list:
|
||||
data = fetch_financial_data(symbol, start_date, end_date)
|
||||
|
||||
if not data:
|
||||
print(f"⚠️ Warning: No data returned for {symbol}. Assigning default score.\n")
|
||||
scores = {screener: 0.25 for category in selected_screeners for screener in selected_screeners[category]}
|
||||
else:
|
||||
scores = {}
|
||||
|
||||
for category, screeners in selected_screeners.items():
|
||||
for screener, threshold in screeners.items():
|
||||
if screener == "EPS_Score":
|
||||
scores[screener] = check_quarterly_earnings(data.get("quarterly_eps", []))
|
||||
elif screener == "Annual_EPS_Score":
|
||||
scores[screener] = check_annual_eps_growth(data.get("annual_eps", []))
|
||||
elif screener == "Sales_Score":
|
||||
scores[screener] = check_sales_growth(data.get("sales_growth", []))
|
||||
elif screener == "ROE_Score":
|
||||
scores[screener] = check_return_on_equity(data.get("roe", []))
|
||||
elif screener == "L_Score":
|
||||
scores[screener] = check_industry_leadership(symbol)
|
||||
print(f"🟢 {symbol} - L_Score: {scores[screener]}")
|
||||
elif screener == "I_Score":
|
||||
scores[screener] = check_institutional_sponsorship(symbol)
|
||||
print(f"🏢 {symbol} - I_Score: {scores[screener]}")
|
||||
|
||||
if isinstance(threshold, (int, float)):
|
||||
scores[screener] = scores[screener] >= threshold
|
||||
|
||||
scores["Total_Score"] = sum(scores.values())
|
||||
append_scores_to_csv(symbol, scores)
|
||||
|
||||
print("✅ Scores saved in data/metrics/stock_scores.csv\n")
|
||||
|
||||
elif choice == "2":
|
||||
min_price, max_price, min_volume = get_scanner_parameters()
|
||||
from screener.t_sunnyband import run_sunny_scanner
|
||||
run_sunny_scanner(min_price, max_price, min_volume)
|
||||
|
||||
elif choice == "3":
|
||||
from trading.main import main as trading_main
|
||||
trading_main()
|
||||
|
||||
elif choice == "4":
|
||||
print("Exiting...")
|
||||
return
|
||||
|
||||
print(f"Processing {len(symbol_list)} stocks within the given date range...\n")
|
||||
|
||||
# 5️⃣ Process each stock symbol
|
||||
for symbol in symbol_list:
|
||||
data = fetch_financial_data(symbol, start_date, end_date)
|
||||
|
||||
if not data:
|
||||
print(f"⚠️ Warning: No data returned for {symbol}. Assigning default score.\n")
|
||||
scores = {screener: 0.25 for category in selected_screeners for screener in selected_screeners[category]}
|
||||
else:
|
||||
scores = {}
|
||||
|
||||
# 6️⃣ Compute scores dynamically based on user selection
|
||||
for category, screeners in selected_screeners.items():
|
||||
for screener, threshold in screeners.items():
|
||||
if screener == "EPS_Score":
|
||||
scores[screener] = check_quarterly_earnings(data.get("quarterly_eps", []))
|
||||
elif screener == "Annual_EPS_Score":
|
||||
scores[screener] = check_annual_eps_growth(data.get("annual_eps", []))
|
||||
elif screener == "Sales_Score":
|
||||
scores[screener] = check_sales_growth(data.get("sales_growth", []))
|
||||
elif screener == "ROE_Score":
|
||||
scores[screener] = check_return_on_equity(data.get("roe", []))
|
||||
elif screener == "L_Score":
|
||||
scores[screener] = check_industry_leadership(symbol) # ✅ NEW: Industry Leadership Calculation
|
||||
print(f"🟢 {symbol} - L_Score: {scores[screener]}") # ✅ DEBUG LOG
|
||||
elif screener == "I_Score":
|
||||
scores[screener] = check_institutional_sponsorship(symbol) # ✅ NEW: Institutional Sponsorship Check
|
||||
print(f"🏢 {symbol} - I_Score: {scores[screener]}") # ✅ DEBUG LOG
|
||||
|
||||
# Apply user-defined threshold if applicable
|
||||
if isinstance(threshold, (int, float)):
|
||||
scores[screener] = scores[screener] >= threshold
|
||||
|
||||
# 7️⃣ Calculate Total Score
|
||||
scores["Total_Score"] = sum(scores.values()) # ✅ NEW: Total Score Calculation
|
||||
|
||||
# 8️⃣ Append results to CSV
|
||||
append_scores_to_csv(symbol, scores)
|
||||
|
||||
print("✅ Scores saved in data/metrics/stock_scores.csv\n")
|
||||
else:
|
||||
print("Invalid choice. Please try again.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user