From cb098278c9eb409a14719bd94dae9ec0b4f0c605 Mon Sep 17 00:00:00 2001 From: Bobby Abellana Date: Mon, 3 Feb 2025 21:53:26 -0800 Subject: [PATCH] C in CANSLIM --- src/db/{queries.py => __init__.py} | 0 src/main.py | 53 ++++++++ .../__init__.py} | 0 src/screener/c_canslim.py | 75 +++++++++++ src/screener/csv_appender.py | 36 +++++ src/screener/data_fetcher.py | 126 ++++++++++++++++++ src/screener/metrics.py | 0 src/screener/screener.py | 0 tests/test_screener.py | 0 9 files changed, 290 insertions(+) rename src/db/{queries.py => __init__.py} (100%) rename src/{patterns/pattern_finder.py => screener/__init__.py} (100%) create mode 100644 src/screener/c_canslim.py create mode 100644 src/screener/csv_appender.py create mode 100644 src/screener/data_fetcher.py delete mode 100644 src/screener/metrics.py delete mode 100644 src/screener/screener.py delete mode 100644 tests/test_screener.py diff --git a/src/db/queries.py b/src/db/__init__.py similarity index 100% rename from src/db/queries.py rename to src/db/__init__.py diff --git a/src/main.py b/src/main.py index e69de29..c50f40f 100644 --- a/src/main.py +++ b/src/main.py @@ -0,0 +1,53 @@ +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_sales_growth, check_return_on_equity +from screener.csv_appender import append_scores_to_csv + +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): ") + + # 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 all stock symbols in the given date range dynamically + 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...") + + # 4️⃣ 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.") + scores = { + "EPS_Score": 0.25, # EPS Growth + "Sales_Score": 0.25, # Sales Growth + "ROE_Score": 0.25 # Return on Equity + } + else: + # Extract relevant fields + quarterly_eps = data.get("eps", []) + sales_growth = data.get("sales_growth", []) + roe = 18 # Placeholder, modify if needed + + # 5️⃣ Compute CANSLIM Scores + scores = { + "EPS_Score": check_quarterly_earnings(quarterly_eps), + "Sales_Score": check_sales_growth(sales_growth), + "ROE_Score": check_return_on_equity(roe) + } + + # 6️⃣ Append results to a **generic** CSV file in `data/metrics/` + append_scores_to_csv(symbol, scores) + + print("Scores saved in data/metrics/stock_scores.csv") + +if __name__ == "__main__": + main() diff --git a/src/patterns/pattern_finder.py b/src/screener/__init__.py similarity index 100% rename from src/patterns/pattern_finder.py rename to src/screener/__init__.py diff --git a/src/screener/c_canslim.py b/src/screener/c_canslim.py new file mode 100644 index 0000000..6057c09 --- /dev/null +++ b/src/screener/c_canslim.py @@ -0,0 +1,75 @@ +def check_quarterly_earnings(quarterly_eps): + """ + Checks 25%+ growth in the most recent quarter and, if possible, + accelerating growth quarter over quarter. + + Returns: + float: Score (1 pass, 0 fail, 0.25 insufficient data). + """ + if not quarterly_eps or len(quarterly_eps) < 2: + return 0.25 # not enough data + + old_eps = quarterly_eps[-2] + new_eps = quarterly_eps[-1] + + if old_eps <= 0: + return 0.25 # can't compute growth properly + + growth = (new_eps - old_eps) / abs(old_eps) * 100 + if growth < 25: + return 0 # fails the 25% growth condition + + # Check acceleration if we have at least 3 quarters + if len(quarterly_eps) >= 3: + older_eps = quarterly_eps[-3] + if older_eps <= 0: + # can't compare properly, but the 25% is met + return 1 + prev_growth = (old_eps - older_eps) / abs(older_eps) * 100 + if growth >= prev_growth: + # accelerating + return 1 + else: + # not accelerating but still >=25% for last quarter + return 1 + + return 1 # only 2 quarters, meets 25% condition + +def check_sales_growth(quarterly_sales): + """ + Checks for 20%-25% or higher sales growth in the most recent quarter. + + Returns: + float: 1 (Pass), 0 (Fail), 0.25 (Insufficient data). + """ + if not quarterly_sales or len(quarterly_sales) < 2: + return 0.25 # Not enough data + + old_sales = quarterly_sales[-2] + new_sales = quarterly_sales[-1] + + # Handle None values before making calculations + if old_sales is None or new_sales is None: + return 0.25 + + if old_sales <= 0: + return 0.25 # Can't compute growth properly + + growth = (new_sales - old_sales) / abs(old_sales) * 100 + + return 1 if growth >= 20 else 0 # Pass if growth is 20%+ + + +def check_return_on_equity(roe): + """ + Checks for >=17% ROE. + + Returns: + float: Score (1 pass, 0 fail, 0.25 insufficient data). + """ + if roe is None: + return 0.25 + + if roe >= 17: + return 1 + return 0 diff --git a/src/screener/csv_appender.py b/src/screener/csv_appender.py new file mode 100644 index 0000000..1373733 --- /dev/null +++ b/src/screener/csv_appender.py @@ -0,0 +1,36 @@ +import csv +import os + +# Get the absolute path to the `data/metrics/` directory (outside `src/`) +BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")) # Go two levels up +METRICS_DIR = os.path.join(BASE_DIR, "data", "metrics") +CSV_FILE = os.path.join(METRICS_DIR, "stock_scores.csv") + +def append_scores_to_csv(symbol, scores): + """ + Append stock analysis scores to a generic CSV file in the `data/metrics` directory. + + Args: + symbol (str): Stock ticker symbol. + scores (dict): Dictionary of metric scores (e.g., C_Score, Sales_Score, etc.). + """ + # Ensure the directory exists + os.makedirs(METRICS_DIR, exist_ok=True) + + # Define the header dynamically based on the provided scores + headers = ["Symbol"] + list(scores.keys()) + + # Check if file exists + file_exists = os.path.isfile(CSV_FILE) + + with open(CSV_FILE, 'a', newline='') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=headers) + + # Write the header if the file is new + if not file_exists: + writer.writeheader() + + # Write stock data + row = {"Symbol": symbol} + row.update(scores) + writer.writerow(row) diff --git a/src/screener/data_fetcher.py b/src/screener/data_fetcher.py new file mode 100644 index 0000000..e22f389 --- /dev/null +++ b/src/screener/data_fetcher.py @@ -0,0 +1,126 @@ +import datetime +from db.db_connection import create_client + +def validate_date_range(start_date, end_date, required_quarters=4): + """ + Ensures we have enough data (e.g., required_quarters) to evaluate + the 'C' in CANSLIM. If the date range is too short, adjust or return a warning. + + Args: + start_date (str or datetime): User-provided start date. + end_date (str or datetime): User-provided end date. + required_quarters (int): Number of quarters needed (default: 4). + + Returns: + (datetime, datetime): Tuple of adjusted (start_date, end_date). + """ + if isinstance(start_date, str): + start_date = datetime.datetime.strptime(start_date, "%Y-%m-%d") + if isinstance(end_date, str): + end_date = datetime.datetime.strptime(end_date, "%Y-%m-%d") + + # Approximate needed delta for required_quarters + needed_days = required_quarters * 91 # ~3 months each + needed_delta = datetime.timedelta(days=needed_days) + + if (end_date - start_date) < needed_delta: + start_date = end_date - needed_delta + print("Warning: Provided date range was too short. Adjusted start_date for enough data.") + + return start_date, end_date + +def fetch_financial_data(symbol, start_date, end_date): + """ + Fetch financial data (EPS, Sales Growth) from stock_financials table + for a given stock symbol within a specific date range. + + Args: + symbol (str): Stock ticker symbol. + start_date (str or datetime): Start date for data retrieval. + end_date (str or datetime): End date for data retrieval. + + Returns: + dict: A dictionary containing calculated EPS and sales growth. + """ + client = create_client() + + # Query stock_financials for revenue, net income, and EPS + query = f""" + SELECT + filing_date, + revenue, + net_income, + diluted_eps -- Using diluted EPS for accuracy + FROM stock_db.stock_financials + WHERE ticker = '{symbol}' + AND filing_date BETWEEN toDate('{start_date}') AND toDate('{end_date}') + AND timeframe = 'quarterly' -- Ensure only quarterly reports are used + ORDER BY filing_date ASC + """ + + result = client.query(query) + + if not result.result_rows: + return {} + + # Extracting data + dates = [] + revenues = [] + net_incomes = [] + eps_values = [] + sales_growth = [] + + for row in result.result_rows: + dates.append(row[0]) + revenues.append(row[1]) + net_incomes.append(row[2]) + eps_values.append(row[3]) # Directly using diluted EPS + + # Calculate Sales Growth (Quarter-over-Quarter) + for i in range(1, len(revenues)): # Start from index 1 since we compare with previous + prev_revenue = revenues[i - 1] + current_revenue = revenues[i] + if prev_revenue > 0: # Avoid division by zero + growth = ((current_revenue - prev_revenue) / prev_revenue) * 100 + else: + growth = None # Not enough data + sales_growth.append(growth) + + # Pad sales_growth list to match length of EPS (since first quarter lacks a previous value) + sales_growth.insert(0, None) # First quarter doesn't have a previous comparison + + return { + "symbol": symbol, + "dates": dates, + "eps": eps_values, + "sales_growth": sales_growth + } + + +def get_stocks_in_time_range(start_date, end_date): + """ + Query ClickHouse for all stock symbols that have data within the given date range. + + Args: + start_date (str or datetime): Start date in 'YYYY-MM-DD' format. + end_date (str or datetime): End date in 'YYYY-MM-DD' format. + + Returns: + list: A list of stock symbols that have data in the specified date range. + """ + if isinstance(start_date, datetime.datetime): + start_date = start_date.strftime("%Y-%m-%d") + if isinstance(end_date, datetime.datetime): + end_date = end_date.strftime("%Y-%m-%d") + + client = create_client() + + query = f""" + SELECT DISTINCT ticker + FROM stock_db.stock_prices_daily + WHERE date BETWEEN toDate('{start_date}') AND toDate('{end_date}') + """ + + result = client.query(query) + + return [row[0] for row in result.result_rows] diff --git a/src/screener/metrics.py b/src/screener/metrics.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/screener/screener.py b/src/screener/screener.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/test_screener.py b/tests/test_screener.py deleted file mode 100644 index e69de29..0000000