شمارش روزهای کاهش پیاپی (streak)

from pyspark.sql import functions as F, Window

w = Window.partitionBy("product_id").orderBy("jour")
w_cumul = w.rowsBetween(Window.unboundedPreceding, 0)

streaks = (
    ventes_jour
    .withColumn("baisse", (F.col("ca") < F.lag("ca").over(w)).cast("int"))
    .withColumn("rupture", F.when(F.col("baisse") == 0, 1).otherwise(0))
    .withColumn("groupe", F.sum("rupture").over(w_cumul))
    .groupBy("product_id", "groupe")
    .agg(F.sum("baisse").alias("jours_de_baisse"),
         F.max("jour").alias("jusqu_au"))
    .filter("jours_de_baisse >= 5")
)
streaks.select("product_id", "jours_de_baisse", "jusqu_au") \
       .orderBy(F.desc("jours_de_baisse")).show(4)