Financial Data: Market Data, ETF Data & Analytics

import math
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window

spark = SparkSession.builder.appName("FinancialRiskAnalytics").getOrCreate()

def compute_daily_vwap(trades_df):
    """Computes daily VWAP per ticker."""
    return trades_df.groupBy("trade_date", "ticker").agg(
        (F.sum(F.col("price") * F.col("volume")) / F.sum("volume")).alias("vwap"),
        F.sum("volume").alias("total_volume"),
        F.first("price").alias("open_price"),
        F.max("price").alias("high_price"),
        F.min("price").alias("low_price"),
        F.last("price").alias("close_price")
    )

def compute_rolling_volatility_and_sharpe(daily_returns_df, risk_free_rate: float = 0.045):
    """
    Computes 30-day rolling annualized volatility and Sharpe ratio.
    Annualization factor for trading days: sqrt(252).
    """
    window_30d = Window.partitionBy("portfolio_id").orderBy("trade_date").rowsBetween(-29, 0)

    metrics_df = daily_returns_df.withColumn(
        "daily_return", 
        (F.col("nav") - F.lag("nav", 1).over(Window.partitionBy("portfolio_id").orderBy("trade_date"))) / 
        F.lag("nav", 1).over(Window.partitionBy("portfolio_id").orderBy("trade_date"))
    ).withColumn(
        "mean_daily_return", F.avg("daily_return").over(window_30d)
    ).withColumn(
        "std_daily_return", F.stddev("daily_return").over(window_30d)
    ).withColumn(
        "annualized_volatility", F.col("std_daily_return") * F.lit(math.sqrt(252))
    ).withColumn(
        "annualized_return", F.col("mean_daily_return") * F.lit(252)
    ).withColumn(
        "sharpe_ratio", 
        (F.col("annualized_return") - F.lit(risk_free_rate)) / F.nullif(F.col("annualized_volatility"), 0)
    )

    return metrics_df