refactor: Move CANSLIM screener to dedicated module
This commit is contained in:
parent
c2feece0d6
commit
ce1db8a8d9
138
src/pages/screener/canslim_screener_page.py
Normal file
138
src/pages/screener/canslim_screener_page.py
Normal file
@ -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")
|
||||
@ -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")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user