From 9b6871d5756b929eee8bb48d020185a9b78ffc4f Mon Sep 17 00:00:00 2001 From: Bobby Abellana Date: Mon, 3 Feb 2025 23:33:42 -0800 Subject: [PATCH] L CANSLIM --- industry_sic_codes.csv | 44 +++++++++++++++++++++ src/main.py | 18 ++++++--- src/screener/csv_appender.py | 73 ++++++++++++++++------------------ src/screener/l_canslim.py | 77 ++++++++++++++++++++++++++++++++++++ src/screener/screeners.py | 3 +- src/screener/user_input.py | 10 ++++- 6 files changed, 177 insertions(+), 48 deletions(-) create mode 100644 industry_sic_codes.csv create mode 100644 src/screener/l_canslim.py diff --git a/industry_sic_codes.csv b/industry_sic_codes.csv new file mode 100644 index 0000000..90f1336 --- /dev/null +++ b/industry_sic_codes.csv @@ -0,0 +1,44 @@ +SIC Code,Office,Industry Title +100,Industrial Applications and Services,AGRICULTURAL PRODUCTION-CROPS +200,Industrial Applications and Services,AGRICULTURAL PROD-LIVESTOCK & ANIMAL SPECIALTIES +700,Industrial Applications and Services,AGRICULTURAL SERVICES +800,Industrial Applications and Services,FORESTRY +900,Industrial Applications and Services,"FISHING, HUNTING AND TRAPPING" +1000,Office of Energy & Transportation,METAL MINING +1040,Office of Energy & Transportation,GOLD AND SILVER ORES +1090,Office of Energy & Transportation,MISCELLANEOUS METAL ORES +1220,Office of Energy & Transportation,BITUMINOUS COAL & LIGNITE MINING +1221,Office of Energy & Transportation,BITUMINOUS COAL & LIGNITE SURFACE MINING +1311,Office of Energy & Transportation,CRUDE PETROLEUM & NATURAL GAS +1381,Office of Energy & Transportation,DRILLING OIL & GAS WELLS +1382,Office of Energy & Transportation,OIL & GAS FIELD EXPLORATION SERVICES +1389,Office of Energy & Transportation,"OIL & GAS FIELD SERVICES, NEC" +1400,Office of Energy & Transportation,MINING & QUARRYING OF NONMETALLIC MINERALS (NO FUELS) +1520,Office of Real Estate & Construction,GENERAL BLDG CONTRACTORS - RESIDENTIAL BLDGS +1531,Office of Real Estate & Construction,OPERATIVE BUILDERS +1540,Office of Real Estate & Construction,GENERAL BLDG CONTRACTORS - NONRESIDENTIAL BLDGS +1600,Office of Real Estate & Construction,HEAVY CONSTRUCTION OTHER THAN BLDG CONST - CONTRACTORS +1623,Office of Real Estate & Construction,"WATER, SEWER, PIPELINE, COMM & POWER LINE CONSTRUCTION" +1700,Office of Real Estate & Construction,CONSTRUCTION - SPECIAL TRADE CONTRACTORS +1731,Office of Real Estate & Construction,ELECTRICAL WORK +2000,Office of Manufacturing,FOOD AND KINDRED PRODUCTS +2011,Office of Manufacturing,MEAT PACKING PLANTS +2013,Office of Manufacturing,SAUSAGES & OTHER PREPARED MEAT PRODUCTS +2015,Office of Manufacturing,POULTRY SLAUGHTERING AND PROCESSING +2020,Office of Manufacturing,DAIRY PRODUCTS +2024,Office of Manufacturing,ICE CREAM & FROZEN DESSERTS +2030,Office of Manufacturing,"CANNED, FROZEN & PRESERVD FRUIT, VEG & FOOD SPECIALTIES" +2033,Office of Manufacturing,"CANNED, FRUITS, VEG, PRESERVES, JAMS & JELLIES" +2040,Office of Manufacturing,GRAIN MILL PRODUCTS +2050,Office of Manufacturing,BAKERY PRODUCTS +2052,Office of Manufacturing,COOKIES & CRACKERS +2060,Office of Manufacturing,SUGAR & CONFECTIONERY PRODUCTS +2070,Office of Manufacturing,FATS & OILS +2080,Office of Manufacturing,BEVERAGES +2082,Office of Manufacturing,MALT BEVERAGES +2086,Office of Manufacturing,BOTTLED & CANNED SOFT DRINKS & CARBONATED WATERS +2090,Office of Manufacturing,MISCELLANEOUS FOOD PREPARATIONS & KINDRED PRODUCTS +2092,Office of Manufacturing,PREPARED FRESH OR FROZEN FISH & SEAFOODS +2100,Office of Manufacturing,TOBACCO PRODUCTS +2111,Office of Manufacturing,CIGARETTES +2200,Office of Manufacturing,TEXTILE MILL PRODUCTS diff --git a/src/main.py b/src/main.py index 7a60e27..1e1e777 100644 --- a/src/main.py +++ b/src/main.py @@ -2,6 +2,7 @@ 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.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 @@ -15,7 +16,8 @@ def main(): 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() # ✅ Now imported from `user_input.py` + 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) @@ -24,14 +26,14 @@ def main(): print("No stocks found within the given date range.") return - print(f"Processing {len(symbol_list)} stocks within the given date range...") + 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.") + 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 = {} @@ -47,15 +49,21 @@ def main(): 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 # Apply user-defined threshold if applicable if isinstance(threshold, (int, float)): scores[screener] = scores[screener] >= threshold - # 7️⃣ Append results to CSV + # 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") + print("✅ Scores saved in data/metrics/stock_scores.csv\n") if __name__ == "__main__": main() diff --git a/src/screener/csv_appender.py b/src/screener/csv_appender.py index fbe7a88..1cb721f 100644 --- a/src/screener/csv_appender.py +++ b/src/screener/csv_appender.py @@ -1,57 +1,50 @@ import csv import os -# Define the directory and generic CSV file path -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") - -# Track all unique screeners used in any stock -ALL_HEADERS = set() +CSV_FILE_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../data/metrics/stock_scores.csv")) def append_scores_to_csv(symbol, scores): """ - Append stock analysis scores to a generic CSV file in `data/metrics` directory. - - Ensures all rows have the same headers, adding new screeners dynamically. - + Appends stock scores to the CSV file dynamically and ensures output order matches CSV. + Args: symbol (str): Stock ticker symbol. - scores (dict): Dictionary of metric scores (e.g., EPS_Score, Sales_Score, etc.). + scores (dict): Dictionary of scores for each screener. """ - # Ensure the directory exists - os.makedirs(METRICS_DIR, exist_ok=True) + file_exists = os.path.exists(CSV_FILE_PATH) - # Compute Total Score - total_score = sum(scores.values()) - scores["Total_Score"] = total_score + # Ensure Total_Score is always included + scores["Total_Score"] = sum(scores.values()) - # Update tracked headers - ALL_HEADERS.update(scores.keys()) + # Read existing headers if the file exists + existing_headers = [] + if file_exists: + with open(CSV_FILE_PATH, mode="r", encoding="utf-8") as file: + reader = csv.reader(file) + existing_headers = next(reader, []) - # Preferred order: "Symbol" first, "Total_Score" last - preferred_order = ["Symbol"] - remaining_headers = sorted([h for h in ALL_HEADERS if h not in preferred_order + ["Total_Score"]]) - sorted_headers = preferred_order + remaining_headers + ["Total_Score"] + # Ensure headers dynamically include all possible screeners, keeping order + new_headers = ["Symbol"] + list(scores.keys()) - # Check if file exists - file_exists = os.path.isfile(CSV_FILE) + # Merge existing headers with new ones, keeping Symbol first and Total_Score last + headers = existing_headers if existing_headers else new_headers + if "L_Score" not in headers: + headers.insert(-1, "L_Score") # Ensure L_Score is before Total_Score - with open(CSV_FILE, 'a', newline='') as csvfile: - writer = csv.DictWriter(csvfile, fieldnames=sorted_headers) + # Ensure order consistency for output + row_data = {header: scores.get(header, 0) for header in headers} + row_data["Symbol"] = symbol # Ensure Symbol is set - # If it's a new file or headers changed, write a new header row - if not file_exists or set(sorted_headers) != set(writer.fieldnames): - csvfile.seek(0) # Move to start of the file - csvfile.truncate() # Clear previous file to rewrite headers + # ✅ Print scores in the exact order they will appear in the CSV + print("\n📊 Score Output (Matching CSV Order):") + print(f"{symbol}: " + ", ".join([f"{key}: {row_data[key]}" for key in headers])) + + # Open CSV and update headers if needed + with open(CSV_FILE_PATH, mode="a", newline="") as file: + writer = csv.DictWriter(file, fieldnames=headers) + + # Write header only if file is new or headers have changed + if not file_exists or set(headers) != set(existing_headers): writer.writeheader() - # Fill missing values with 0.25 (default) - row = {"Symbol": symbol} - for header in sorted_headers: - if header == "Symbol": - row[header] = symbol - else: - row[header] = scores.get(header, 0.25) # Use default if missing - - writer.writerow(row) + writer.writerow(row_data) diff --git a/src/screener/l_canslim.py b/src/screener/l_canslim.py new file mode 100644 index 0000000..a3b42b9 --- /dev/null +++ b/src/screener/l_canslim.py @@ -0,0 +1,77 @@ +import csv +import os +from db.db_connection import create_client + +# Load SIC Industry Data +SIC_LOOKUP = {} + +def load_sic_data(): + """Loads SIC Code data into a dictionary from the new CSV location above `src/`.""" + global SIC_LOOKUP + script_dir = os.path.dirname(os.path.abspath(__file__)) # Get the directory of l_canslim.py + project_root = os.path.abspath(os.path.join(script_dir, "../../")) # Go up to project root + sic_file = os.path.join(project_root, "industry_sic_codes.csv") # Adjust path + + if not os.path.exists(sic_file): + raise FileNotFoundError(f"Error: SIC Code CSV file not found at {sic_file}") + + with open(sic_file, mode="r", encoding="utf-8") as file: + reader = csv.DictReader(file) + for row in reader: + sic_code = row["SIC Code"].strip() # Match CSV header exactly + industry = row["Industry Title"].strip() # Match CSV header exactly + SIC_LOOKUP[sic_code] = industry + +# Ensure SIC data is loaded at module import +load_sic_data() + +def check_industry_leadership(symbol): + """ + Determines if a stock is a leader in its industry group. + + Criteria: + - Relative Strength (RS) Rating should be **80 or higher**. + - Stock should be in the top-performing industry groups (top 40-50). + - Stock should be trading near its **52-week high**. + + Args: + symbol (str): Stock ticker symbol. + + Returns: + float: 1 (Pass), 0 (Fail), 0.25 (Insufficient Data). + """ + client = create_client() + + query = f""" + SELECT + ticker, + sic, + relative_strength, + high_52_week, + current_price + FROM stock_db.stock_indicators + WHERE ticker = '{symbol}' + """ + + result = client.query(query) + + if not result.result_rows: + return 0.25 # Not enough data + + _, sic, rs_rating, high_52_week, current_price = result.result_rows[0] + + if not sic or str(sic) not in SIC_LOOKUP: + return 0.25 # No SIC industry data available + + # Ensure RS rating is valid + if rs_rating is None or high_52_week is None or current_price is None: + return 0.25 # Missing necessary data + + # Industry Leader Criteria + passes_rs = rs_rating >= 80 + near_high = current_price >= (high_52_week * 0.90) # Within 10% of 52-week high + + if passes_rs and near_high: + return 1 # Stock is an industry leader + else: + return 0 # Stock is lagging diff --git a/src/screener/screeners.py b/src/screener/screeners.py index 2a6c4d6..acae5b3 100644 --- a/src/screener/screeners.py +++ b/src/screener/screeners.py @@ -4,7 +4,8 @@ SCREENERS = { "EPS_Score": "Checks quarterly EPS growth", "Annual_EPS_Score": "Checks 3-year annual EPS growth", "Sales_Score": "Checks quarterly sales growth", - "ROE_Score": "Checks return on equity" + "ROE_Score": "Checks return on equity", + "L_Score": "Checks if the stock is a leader in its industry" # ✅ NEW: Added L_Score }, "Volume-Based": { "Volume_Oscillator_Score": "Checks for unusual volume surges", diff --git a/src/screener/user_input.py b/src/screener/user_input.py index 8057f3d..9a39e7c 100644 --- a/src/screener/user_input.py +++ b/src/screener/user_input.py @@ -21,9 +21,9 @@ def get_user_screener_selection(): for category in selected_categories: print(f"\nCategory: {category}") use_defaults = input(f"Use default settings for {category}? (y/n): ").strip().lower() - + selected_screeners[category] = {} - + for screener, description in SCREENERS[category].items(): if use_defaults == "y": selected_screeners[category][screener] = "default" @@ -31,4 +31,10 @@ def get_user_screener_selection(): custom_value = input(f"{screener} ({description}) - Enter custom threshold or press Enter to use default: ").strip() selected_screeners[category][screener] = float(custom_value) if custom_value else "default" + # ✅ Ensure L_Score is added if Fundamentals is selected + if "Fundamentals" in selected_screeners and "L_Score" not in selected_screeners["Fundamentals"]: + selected_screeners["Fundamentals"]["L_Score"] = "default" # ✅ Ensure L_Score is included + + print(f"\n✅ Selected Screeners: {selected_screeners}\n") # ✅ DEBUG LOG + return selected_screeners