Modified python code files

Author: dimtics

pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     "polars>=1.12.0",
     "pyarrow>=18.0.0",
     "pydantic>=2.9.2",
+    "python-dotenv>=1.0.1",
     "pyyaml>=6.0.2",
 ]
 

src/stock_valuation_app/__init__.py

@@ -1,2 +1,9 @@
-def main() -> None:
-    print("Hello from stock-valuation-app!")
+from fastapi import FastAPI
+from .api.routes import router
+from .ui.dashboard import app as dash_app
+
+app = FastAPI()
+app.include_router(router)
+
+# Mount Dash app
+app.mount("/dashboard", dash_app.server)

src/stock_valuation_app/api/routes.py

@@ -0,0 +1,33 @@
+from fastapi import APIRouter, Depends
+from ..data.fmp_client import FMPClient
+from ..services.valuation import (
+    calculate_growth_rates,
+    is_quality_dividend_growth_stock,
+    is_undervalued,
+)
+
+router = APIRouter()
+
+
+async def get_fmp_client():
+    return FMPClient("YOUR_API_KEY")
+
+
+@router.get("/stock/{symbol}")
+async def get_stock_valuation(
+    symbol: str, fmp_client: FMPClient = Depends(get_fmp_client)
+):
+    financial_data = await fmp_client.get_financial_data(symbol)
+    growth_rates = calculate_growth_rates(financial_data)
+
+    # Assuming we have a way to get the current P/E ratio
+    current_pe = 15  # This should be fetched from the API
+
+    return {
+        "symbol": symbol,
+        "growth_rates": growth_rates,
+        "is_quality_dividend_growth_stock": is_quality_dividend_growth_stock(
+            growth_rates
+        ),
+        "is_undervalued": is_undervalued(growth_rates, current_pe),
+    }

src/stock_valuation_app/config.yml

@@ -1,5 +1,8 @@
 api:
   base_url: "https://financialmodelingprep.com/api/v3"
+  annual_ratios: "ratios/"
+  annual_financial_growth: "financial-growth/"
+  rating: "rating/"
 
 valuation:
   default_pe_ratio: 15
@@ -15,5 +18,8 @@ database:
   file: "stock_data.duckdb"
 
 logging:
-  level: "INFO"
-  file: "app.log"
+  level: INFO
+  file: app.log
+  filemode: "w"
+  format: "%(asctime)s - %(levelname)s - %(message)s"
+  datefmt: "%Y-%m-%d %H:%M:%S"

src/stock_valuation_app/data/fmp_client.py

@@ -1,25 +1,33 @@
+from dataclasses import dataclass, field
+from urllib.parse import urljoin
 import httpx
+from typing import Any, Optional
 from pydantic import BaseModel
+from stock_valuation_app.utils import utils
 
 
-class FinancialData(BaseModel):
-    revenue: float
-    netIncome: float
-    freeCashFlow: float
-    dividendsPaid: float
+@dataclass
+class FMPClient:
+    """A client for interacting with the Financial Modeling Prep API."""
+    base_url: str = field(default_factory=utils.get_base_url)
+    api_key: str = field(default_factory=utils.get_api_key)
 
 
-class FMPClient:
-    def __init__(self, api_key: str):
-        self.base_url = "https://financialmodelingprep.com/api/v3"
-        self.api_key = api_key
+    async def fetch_data(self, endpoint: str, symbol: str, period: Optional[str] = "annual"):  # params: dict[str, Any]
+        """Fetch data from the Financial Modeling Prep API."""
+        base_endpoint = urljoin(self.base_url, endpoint)
+        if period is None:
+            url = f"{base_endpoint}/{symbol}?apikey={self.api_key}"
+        else:
+            url = f"{base_endpoint}/{symbol}?period={period}&apikey={self.api_key}"
 
-    async def get_financial_data(
-        self, symbol: str, limit: int = 10
-    ) -> list[FinancialData]:
-        url = f"{self.base_url}/income-statement/{symbol}?limit={limit}&apikey={self.api_key}"
         async with httpx.AsyncClient() as client:
             response = await client.get(url)
             response.raise_for_status()
             data = response.json()
-            return [FinancialData(**item) for item in data]
+            return data
+
+api_client = FMPClient()
+rating_endpoint = utils.get_endpoint("rating")
+data = api_client.fetch_data(rating_endpoint, "AAPL")
+print(data)

src/stock_valuation_app/main.py

@@ -0,0 +1,36 @@
+import polars as pl
+from ..data.fmp_client import FinancialData
+
+
+def calculate_growth_rates(data: list[FinancialData]) -> dict[str, dict[str, float]]:
+    df = pl.DataFrame([d.dict() for d in data])
+
+    metrics = ["revenue", "netIncome", "freeCashFlow", "dividendsPaid"]
+    periods = [3, 5, 10]
+
+    results = {}
+    for metric in metrics:
+        metric_results = {}
+        for period in periods:
+            if len(df) >= period:
+                growth_rate = (df[metric][0] / df[metric][period - 1]) ** (
+                    1 / period
+                ) - 1
+                metric_results[f"{period}y"] = growth_rate
+        results[metric] = metric_results
+
+    return results
+
+
+def is_quality_dividend_growth_stock(growth_rates: dict[str, dict[str, float]]) -> bool:
+    revenue_growth = growth_rates["revenue"]["5y"]
+    dividend_growth = growth_rates["dividendsPaid"]["5y"]
+    return revenue_growth > 0.05 and dividend_growth > 0.05
+
+
+def is_undervalued(
+    growth_rates: dict[str, dict[str, float]], current_pe: float
+) -> bool:
+    earnings_growth = growth_rates["netIncome"]["5y"]
+    fair_pe = earnings_growth * 100  # PEG ratio of 1
+    return current_pe < fair_pe

src/stock_valuation_app/services/valuation.py

@@ -0,0 +1,54 @@
+import dash
+from dash import dcc, html
+from dash.dependencies import Input, Output
+import plotly.graph_objs as go
+import httpx
+
+app = dash.Dash(__name__)
+
+app.layout = html.Div(
+    [
+        html.H1("Stock Valuation Dashboard"),
+        dcc.Input(id="stock-input", type="text", placeholder="Enter stock symbol"),
+        html.Button("Analyze", id="analyze-button"),
+        html.Div(id="valuation-output"),
+        dcc.Graph(id="growth-chart"),
+    ]
+)
+
+
+@app.callback(
+    [Output("valuation-output", "children"), Output("growth-chart", "figure")],
+    [Input("analyze-button", "n_clicks")],
+    [dash.dependencies.State("stock-input", "value")],
+)
+def update_valuation(n_clicks, symbol):
+    if n_clicks is None or not symbol:
+        return dash.no_update, dash.no_update
+
+    # Fetch data from our FastAPI endpoint
+    response = httpx.get(f"http://localhost:8000/stock/{symbol}")
+    data = response.json()
+
+    # Create valuation output
+    valuation_output = [
+        html.P(
+            f"Is quality dividend growth stock: {data['is_quality_dividend_growth_stock']}"
+        ),
+        html.P(f"Is undervalued: {data['is_undervalued']}"),
+    ]
+
+    # Create growth chart
+    traces = []
+    for metric, rates in data["growth_rates"].items():
+        trace = go.Bar(x=list(rates.keys()), y=list(rates.values()), name=metric)
+        traces.append(trace)
+
+    layout = go.Layout(title="Growth Rates", barmode="group")
+    figure = go.Figure(data=traces, layout=layout)
+
+    return valuation_output, figure
+
+
+if __name__ == "__main__":
+    app.run_server(debug=True)

src/stock_valuation_app/ui/dashboard.py

@@ -0,0 +1,45 @@
+import os
+from pathlib import Path
+from dotenv import load_dotenv
+import yaml
+
+# Load environment variables from .env file
+load_dotenv()
+# Define the path to the config.yml file
+config_file = Path("../config.yml")
+
+def load_config():
+    """Load configuration from a YAML file."""
+    with open(f"{config_file}", "r", encoding="utf-8") as file:
+        config = yaml.safe_load(file)
+    return config
+
+
+def get_section_config(section: str):
+    """Get configuration for a specific section."""
+    match section:
+        case "api":
+            return load_config().get("api")
+        case "valuation":
+            return load_config().get("valuation")
+        case "dashboard":
+            return load_config().get("dashboard")
+        case "database":
+            return load_config().get("database")
+        case "logging":
+            return load_config().get("logging")
+        case _:
+            raise ValueError(f"Invalid section: {section}")
+
+def get_endpoint(url: str) -> str:
+    """Get endpoints from the configuration."""
+    api_config = get_section_config("api")
+    return api_config.get(f"{url}")
+
+def get_base_url() -> str:
+    """Get the base URL from the configuration."""
+    return get_endpoint("base_url")
+
+def get_api_key() -> str:
+    """Get the API key from the .env file."""
+    return os.getenv("FMP_API_KEY", "")