From ce1db8a8d9df2c4982a3c301a0661a30fcc097fb Mon Sep 17 00:00:00 2001 From: "Bobby (aider)" Date: Wed, 12 Feb 2025 17:00:08 -0800 Subject: [PATCH] refactor: Move CANSLIM screener to dedicated module --- src/pages/screener/canslim_screener_page.py | 138 ++++++++++++++++++++ src/streamlit_app.py | 134 +------------------ 2 files changed, 139 insertions(+), 133 deletions(-) create mode 100644 src/pages/screener/canslim_screener_page.py diff --git a/src/pages/screener/canslim_screener_page.py b/src/pages/screener/canslim_screener_page.py new file mode 100644 index 0000000..6b1adac --- /dev/null +++ b/src/pages/screener/canslim_screener_page.py @@ -0,0 +1,138 @@ +import streamlit as st +import pandas as pd +from datetime import datetime +from screener.canslim_controller import run_canslim_screener +from db.db_connection import create_client + +def load_scanner_reports(): + """Load and return available scanner reports""" + import os + import pandas as pd + from datetime import datetime + + reports = [] + reports_dir = "reports" + + if os.path.exists(reports_dir): + for file in os.listdir(reports_dir): + if file.endswith(".csv"): + file_path = os.path.join(reports_dir, file) + # Get file creation time + created = datetime.fromtimestamp(os.path.getctime(file_path)) + reports.append({ + 'name': file, + 'path': file_path, + 'created': created + }) + + # Sort by creation time, newest first + return sorted(reports, key=lambda x: x['created'], reverse=True) + +def canslim_screener_page(): + st.header("CANSLIM Screener") + + # Create tabs for scanner and reports + scanner_tab, reports_tab = st.tabs(["Run Scanner", "View Reports"]) + + with scanner_tab: + # Date range selection + col1, col2 = st.columns(2) + with col1: + start_date = st.date_input("Start Date") + with col2: + end_date = st.date_input("End Date") + + # CANSLIM criteria selection + st.subheader("Select Screening Criteria") + + c_criteria = st.expander("Current Quarterly Earnings (C)") + with c_criteria: + eps_threshold = st.slider("EPS Growth Threshold (%)", 0, 100, 25) + sales_threshold = st.slider("Sales Growth Threshold (%)", 0, 100, 25) + roe_threshold = st.slider("ROE Threshold (%)", 0, 50, 17) + + a_criteria = st.expander("Annual Earnings Growth (A)") + with a_criteria: + annual_eps_threshold = st.slider("Annual EPS Growth Threshold (%)", 0, 100, 25) + + l_criteria = st.expander("Industry Leadership (L)") + with l_criteria: + use_l = st.checkbox("Check Industry Leadership", value=True) + l_threshold = st.slider("Industry Leadership Score Threshold", 0.0, 1.0, 0.7) + + i_criteria = st.expander("Institutional Sponsorship (I)") + with i_criteria: + use_i = st.checkbox("Check Institutional Sponsorship", value=True) + i_threshold = st.slider("Institutional Sponsorship Score Threshold", 0.0, 1.0, 0.7) + + if st.button("Run CANSLIM Screener"): + with st.spinner("Running CANSLIM screener..."): + try: + # Prepare selected screeners dictionary + selected_screeners = { + "C": { + "EPS_Score": eps_threshold / 100, + "Sales_Score": sales_threshold / 100, + "ROE_Score": roe_threshold / 100 + }, + "A": { + "Annual_EPS_Score": annual_eps_threshold / 100 + } + } + + if use_l: + selected_screeners["L"] = {"L_Score": l_threshold} + if use_i: + selected_screeners["I"] = {"I_Score": i_threshold} + + # Convert dates to strings for the screener + start_str = start_date.strftime("%Y-%m-%d") + end_str = end_date.strftime("%Y-%m-%d") + + # Run the screener + st.session_state.screener_params = { + "start_date": start_str, + "end_date": end_str, + "selected_screeners": selected_screeners + } + + # Modify run_canslim_screener to accept parameters + run_canslim_screener() + st.success("Screening complete! Check the Reports tab for results.") + + except Exception as e: + st.error(f"Error running CANSLIM screener: {str(e)}") + + with reports_tab: + st.subheader("CANSLIM Reports") + + reports = load_scanner_reports() + if reports: + # Create a selectbox to choose the report + selected_report = st.selectbox( + "Select Report", + options=reports, + format_func=lambda x: f"{x['name']} ({x['created'].strftime('%Y-%m-%d %H:%M')})", + key="canslim_scanner_report" + ) + + if selected_report: + try: + # Load and display the CSV + df = pd.read_csv(selected_report['path']) + + # Add download button + st.download_button( + label="Download Report", + data=df.to_csv(index=False), + file_name=selected_report['name'], + mime='text/csv' + ) + + # Display the dataframe + st.dataframe(df) + + except Exception as e: + st.error(f"Error loading report: {str(e)}") + else: + st.info("No CANSLIM reports found") diff --git a/src/streamlit_app.py b/src/streamlit_app.py index 61a5988..34a4a3e 100644 --- a/src/streamlit_app.py +++ b/src/streamlit_app.py @@ -18,7 +18,7 @@ from trading.trading_plan import ( from trading.position_calculator import PositionCalculator from pages.rules.strategy_guide_page import strategy_guide_page from screener.scanner_controller import run_technical_scanner -from screener.canslim_controller import run_canslim_screener +from pages.screener.canslim_screener_page import canslim_screener_page from trading.portfolio import Portfolio, Position from trading.position_calculator import PositionCalculator import plotly.graph_objects as go @@ -29,29 +29,6 @@ def init_session_state(): if 'page' not in st.session_state: st.session_state.page = 'Trading Journal' -def load_scanner_reports(): - """Load and return available scanner reports""" - import os - import pandas as pd - from datetime import datetime - - reports = [] - reports_dir = "reports" - - if os.path.exists(reports_dir): - for file in os.listdir(reports_dir): - if file.endswith(".csv"): - file_path = os.path.join(reports_dir, file) - # Get file creation time - created = datetime.fromtimestamp(os.path.getctime(file_path)) - reports.append({ - 'name': file, - 'path': file_path, - 'created': created - }) - - # Sort by creation time, newest first - return sorted(reports, key=lambda x: x['created'], reverse=True) def technical_scanner_page(): @@ -165,115 +142,6 @@ def technical_scanner_page(): else: st.info("No scanner reports found") -def canslim_screener_page(): - st.header("CANSLIM Screener") - - # Create tabs for scanner and reports - scanner_tab, reports_tab = st.tabs(["Run Scanner", "View Reports"]) - - with scanner_tab: - # Date range selection - col1, col2 = st.columns(2) - with col1: - start_date = st.date_input("Start Date") - with col2: - end_date = st.date_input("End Date") - - # CANSLIM criteria selection - st.subheader("Select Screening Criteria") - - c_criteria = st.expander("Current Quarterly Earnings (C)") - with c_criteria: - eps_threshold = st.slider("EPS Growth Threshold (%)", 0, 100, 25) - sales_threshold = st.slider("Sales Growth Threshold (%)", 0, 100, 25) - roe_threshold = st.slider("ROE Threshold (%)", 0, 50, 17) - - a_criteria = st.expander("Annual Earnings Growth (A)") - with a_criteria: - annual_eps_threshold = st.slider("Annual EPS Growth Threshold (%)", 0, 100, 25) - - l_criteria = st.expander("Industry Leadership (L)") - with l_criteria: - use_l = st.checkbox("Check Industry Leadership", value=True) - l_threshold = st.slider("Industry Leadership Score Threshold", 0.0, 1.0, 0.7) - - i_criteria = st.expander("Institutional Sponsorship (I)") - with i_criteria: - use_i = st.checkbox("Check Institutional Sponsorship", value=True) - i_threshold = st.slider("Institutional Sponsorship Score Threshold", 0.0, 1.0, 0.7) - - if st.button("Run CANSLIM Screener"): - with st.spinner("Running CANSLIM screener..."): - try: - # Prepare selected screeners dictionary - selected_screeners = { - "C": { - "EPS_Score": eps_threshold / 100, - "Sales_Score": sales_threshold / 100, - "ROE_Score": roe_threshold / 100 - }, - "A": { - "Annual_EPS_Score": annual_eps_threshold / 100 - } - } - - if use_l: - selected_screeners["L"] = {"L_Score": l_threshold} - if use_i: - selected_screeners["I"] = {"I_Score": i_threshold} - - # Convert dates to strings for the screener - start_str = start_date.strftime("%Y-%m-%d") - end_str = end_date.strftime("%Y-%m-%d") - - # Run the screener - st.session_state.screener_params = { - "start_date": start_str, - "end_date": end_str, - "selected_screeners": selected_screeners - } - - # Modify run_canslim_screener to accept parameters - run_canslim_screener() - st.success("Screening complete! Check the Reports tab for results.") - - except Exception as e: - st.error(f"Error running CANSLIM screener: {str(e)}") - - with reports_tab: - st.subheader("CANSLIM Reports") - - # Use the same report loading function but look in a CANSLIM-specific directory - reports = load_scanner_reports() # You might want to modify this to look in a CANSLIM directory - if reports: - # Create a selectbox to choose the report - selected_report = st.selectbox( - "Select Report", - options=reports, - format_func=lambda x: f"{x['name']} ({x['created'].strftime('%Y-%m-%d %H:%M')})", - key="canslim_scanner_report" - ) - - if selected_report: - try: - # Load and display the CSV - df = pd.read_csv(selected_report['path']) - - # Add download button - st.download_button( - label="Download Report", - data=df.to_csv(index=False), - file_name=selected_report['name'], - mime='text/csv' - ) - - # Display the dataframe - st.dataframe(df) - - except Exception as e: - st.error(f"Error loading report: {str(e)}") - else: - st.info("No CANSLIM reports found") def trading_system_page(): st.header("Trading System")