stock_system/src/screener/csv_appender.py

53 lines
2.0 KiB
Python

import csv
import os
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):
"""
Appends stock scores to the CSV file dynamically and ensures output order matches CSV.
Args:
symbol (str): Stock ticker symbol.
scores (dict): Dictionary of scores for each screener.
"""
file_exists = os.path.exists(CSV_FILE_PATH)
# Ensure Total_Score is always included
scores["Total_Score"] = sum(scores.values())
# 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, [])
# Ensure headers dynamically include all possible screeners, keeping order
new_headers = ["Symbol"] + list(scores.keys())
# 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
if "I_Score" not in headers:
headers.insert(-1, "I_Score") # Ensure I_Score is before Total_Score
# Ensure order consistency for output
row_data = {header: scores.get(header, 0) for header in headers}
row_data["Symbol"] = symbol # Ensure Symbol is set
# ✅ 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()
writer.writerow(row_data)