CNAM-164 Fixed purchase-based cumulative exposures

danielpes · danielpes · commit b062b0677a95 · 2017-02-10T10:03:31.000+01:00
Now it discretizes the number of purchases

CNAM-164 Removed untested file
diff --git a/src/main/resources/config/filtering-default.conf b/src/main/resources/config/filtering-default.conf
@@ -57,7 +57,8 @@ default = {
       window = 1 #Number of months to quantile.
       start_threshold = 6 #Number of months within which more than one purchases have to made
       end_threshold = 4 #Number of months during which no purchases of the particular molecule have to be made
-      dosage_level_intervals = [0,100,200,300,400,500] #list of consumption levels in mg / put only 0 when we want all the weights to 1
+      dosage_level_intervals = [0, 100, 200, 300, 400, 500] #list of consumption levels in mg / put only 0 when we want all the weights to 1
+      purchase_intervals = [0, 3, 5] #list of consumption levels in mg / put only 0 when we want all the weights to 1
     }
   }
 
diff --git a/src/main/scala/fr/polytechnique/cmap/cnam/filtering/FilteringConfig.scala b/src/main/scala/fr/polytechnique/cmap/cnam/filtering/FilteringConfig.scala
@@ -154,6 +154,7 @@ object FilteringConfig {
     cumulativeExposureWindow = conf.getInt("exposures.cumulative.window"),
     cumulativeStartThreshold = conf.getInt("exposures.cumulative.start_threshold"),
     cumulativeEndThreshold = conf.getInt("exposures.cumulative.end_threshold"),
-    dosageLevelIntervals = conf.getIntList("exposures.cumulative.dosage_level_intervals").asScala.toList.map(x => x.toInt)
+    dosageLevelIntervals = conf.getIntList("exposures.cumulative.dosage_level_intervals").asScala.map(_.toInt).toList,
+    purchaseIntervals = conf.getIntList("exposures.cumulative.purchase_intervals").asScala.map(_.toInt).toList
   )
 }
diff --git a/src/main/scala/fr/polytechnique/cmap/cnam/filtering/cox/NewCoxMain.scala b/src/main/scala/fr/polytechnique/cmap/cnam/filtering/cox/NewCoxMain.scala
diff --git a/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/DosageBasedWeightAgg.scala b/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/DosageBasedWeightAgg.scala
@@ -16,7 +16,9 @@ class DosageBasedWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data) {
     val window = Window.partitionBy("patientID", "eventId")
     val finalWindow = Window.partitionBy("patientID", "eventId", "weight")
 
-    val getLevel = udf{(Quantity:Double) => dosageLevelIntervals.filter(x => x <= Quantity).size}
+    val getLevel = udf {
+      (weight: Double) => dosageLevelIntervals.count(_ <= weight).toDouble
+    }
 
     data
       .withColumn("exposureStart", col("start")) // temporary (todo: "first-only" feature in unlimitedPeriodAdder)
@@ -29,7 +31,8 @@ class DosageBasedWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data) {
       cumWindow: Option[Int] = None,
       cumStartThreshold: Option[Int] = None,
       cumEndThreshold: Option[Int] = None,
-      dosageLevelIntervals: Option[List[Int]]): DataFrame = {
+      dosageLevelIntervals: Option[List[Int]],
+      purchaseIntervals: Option[List[Int]]): DataFrame = {
 
     aggregateWeightImpl(dosageLevelIntervals.get)
   }
diff --git a/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/ExposuresConfig.scala b/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/ExposuresConfig.scala
@@ -16,7 +16,8 @@ case class ExposuresConfig(
     cumulativeExposureWindow: Int,
     cumulativeStartThreshold: Int,
     cumulativeEndThreshold: Int,
-    dosageLevelIntervals: List[Int])
+    dosageLevelIntervals: List[Int],
+    purchaseIntervals: List[Int])
 
 object ExposuresConfig {
   // todo: Remove filters from ExposuresConfig pipeline
diff --git a/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/ExposuresTransformer.scala b/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/ExposuresTransformer.scala
@@ -37,7 +37,14 @@ class ExposuresTransformer(config: ExposuresConfig)
       .where(col("category") === "molecule")
       .withStartEnd(config.minPurchases,config.startDelay,config.purchasesWindow)
       .where(col("exposureStart") !== col("exposureEnd")) // This also removes rows where exposureStart = null
-      .aggregateWeight(Some(config.studyStart), Some(config.cumulativeExposureWindow), Some(config.cumulativeStartThreshold), Some(config.cumulativeEndThreshold), Some(config.dosageLevelIntervals))
+      .aggregateWeight(
+        Some(config.studyStart),
+        Some(config.cumulativeExposureWindow),
+        Some(config.cumulativeStartThreshold),
+        Some(config.cumulativeEndThreshold),
+        Some(config.dosageLevelIntervals),
+        Some(config.purchaseIntervals)
+      )
       .dropDuplicates(Seq("patientID", "eventID", "exposureStart", "exposureEnd", "weight"))
       .select(outputColumns: _*)
       .as[FlatEvent]
diff --git a/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/NonCumulativeWeightAgg.scala b/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/NonCumulativeWeightAgg.scala
@@ -13,7 +13,8 @@ class NonCumulativeWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data)
       cumWindow: Option[Int],
       cumStartThreshold: Option[Int],
       cumEndThreshold: Option[Int],
-      dosageLevelIntervals: Option[List[Int]]): DataFrame = this.aggregateWeightImpl
+      dosageLevelIntervals: Option[List[Int]],
+      purchaseIntervals: Option[List[Int]]): DataFrame = this.aggregateWeightImpl
 
-  def aggregateWeight: DataFrame = aggregateWeight(None, None, None, None, None)
+  def aggregateWeight: DataFrame = aggregateWeight(None, None, None, None, None, None)
 }
diff --git a/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/PurchaseBasedWeightAgg.scala b/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/PurchaseBasedWeightAgg.scala
@@ -10,40 +10,31 @@ import fr.polytechnique.cmap.cnam.utilities.functions.makeTS
 
 class PurchaseBasedWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data) {
 
-  private def aggregateWeightImpl(studyStart: Timestamp, cumWindow: Int): DataFrame = {
+  private def aggregateWeightImpl(purchaseIntervals: List[Int]): DataFrame = {
 
-    val outputColumns: Seq[Column] = data.columns.map(col) :+ col("weight")
     val window = Window.partitionBy("patientID", "eventId")
-    val windowCumulativeExposure = window.partitionBy("patientID", "eventId", "exposureStart")
+    val finalWindow = Window.partitionBy("patientID", "eventId", "weight")
 
-    def normalizeStartMonth(start: Column) = {
-      floor(months_between(start, lit(studyStart)) / cumWindow).cast(IntegerType)
+    val getLevel = udf {
+      (weight: Double) => purchaseIntervals.count(_ <= weight).toDouble
     }
 
-    val normalizeStart = udf(
-      (normalizedMonth: Int) => {
-        val cal: Calendar = Calendar.getInstance()
-        cal.setTime(studyStart)
-        cal.add(Calendar.MONTH, normalizedMonth * cumWindow)
-        makeTS(cal.get(Calendar.YEAR), cal.get(Calendar.MONTH) + 1, 1)
-      }
-    )
-
     data
-      .withColumn("normalizedMonth", normalizeStartMonth(col("start")))
-      .withColumn("exposureStart", normalizeStart(col("normalizedMonth")))
-      .withColumn("weight", row_number().over(window.orderBy("start")))
-      .withColumn("weight", max("weight").over(windowCumulativeExposure).cast(DoubleType))
-      .select(outputColumns: _*)
+      .withColumn("exposureStart", col("start")) // temporary (todo: "first-only" feature in unlimitedPeriodAdder)
+      .withColumn("weight", (sum(lit(1.0)).over(window.orderBy("exposureStart"))))
+      .withColumn("weight", getLevel(col("weight")))
+      .withColumn("exposureStart", min("exposureStart").over(finalWindow))
   }
 
   def aggregateWeight(
-      studyStart: Option[Timestamp],
-      cumWindow: Option[Int],
-      cumStartThreshold: Option[Int] = None,
-      cumEndThreshold: Option[Int] = None,
-      dosageLevelIntervals: Option[List[Int]]= None): DataFrame = {
-
-    this.aggregateWeightImpl(studyStart.get, cumWindow.get)
+    studyStart: Option[Timestamp],
+    cumWindow: Option[Int],
+    cumStartThreshold: Option[Int] = None,
+    cumEndThreshold: Option[Int] = None,
+    dosageLevelIntervals: Option[List[Int]]= None,
+    purchaseIntervals: Option[List[Int]]): DataFrame = {
+
+    this.aggregateWeightImpl(purchaseIntervals.get)
   }
 }
+
diff --git a/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/TimeBasedWeightAgg.scala b/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/TimeBasedWeightAgg.scala
@@ -12,7 +12,8 @@ class TimeBasedWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data) {
       cumWindow: Option[Int],
       cumStartThreshold: Option[Int],
       cumEndThreshold: Option[Int],
-      dosageLevelIntervals: Option[List[Int]]): DataFrame = {
+      dosageLevelIntervals: Option[List[Int]],
+      purchaseIntervals: Option[List[Int]]): DataFrame = {
 
     val window = Window.partitionBy("patientID", "eventId").orderBy("exposureStart", "exposureEnd")
     data
@@ -22,5 +23,5 @@ class TimeBasedWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data) {
       .withColumn("exposureEnd", col("followUpEnd"))
   }
 
-  def aggregateWeight: DataFrame = aggregateWeight(None, None, None, None, None)
+  def aggregateWeight: DataFrame = aggregateWeight(None, None, None, None, None, None)
 }
diff --git a/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/WeightAggregator.scala b/src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/WeightAggregator.scala
@@ -19,10 +19,13 @@ trait WeightAggregator {
 
 abstract class WeightAggregatorImpl(data: DataFrame) {
 
+  // todo: refactor the parametrization (maybe passing a single config object).
+  // The current approach is not maintainable nor scalable
   def aggregateWeight(
-      studyStart: Option[Timestamp],
-      cumWindow: Option[Int],
-      cumStartThreshold: Option[Int],
-      cumEndThreshold: Option[Int],
-      dosageLevelIntervals: Option[List[Int]]): DataFrame
+      studyStart: Option[Timestamp] = None,
+      cumWindow: Option[Int] = None,
+      cumStartThreshold: Option[Int] = None,
+      cumEndThreshold: Option[Int] = None,
+      dosageLevelIntervals: Option[List[Int]] = None,
+      purchaseIntervals: Option[List[Int]] = None): DataFrame
 }
diff --git a/src/test/scala/fr/polytechnique/cmap/cnam/filtering/cox/CoxMainSuite.scala b/src/test/scala/fr/polytechnique/cmap/cnam/filtering/cox/CoxMainSuite.scala
@@ -139,15 +139,12 @@ class CoxMainSuite extends SharedContext {
       .as[FlatEvent]
 
     val expectedResult = Seq(
-      CoxFeature("Patient_A", 1, 566, "45-49", 0, 1, 0, 0, 0, 0, 2, 1, 0),
-      CoxFeature("Patient_A", 1, 566, "45-49", 1, 2, 0, 0, 0, 0, 2, 2, 0),
-      CoxFeature("Patient_A", 1, 566, "45-49", 2, 4, 0, 0, 0, 0, 3, 3, 0),
-      CoxFeature("Patient_A", 1, 566, "45-49", 4, 5, 0, 0, 0, 0, 4, 3, 0),
-      CoxFeature("Patient_A", 1, 566, "45-49", 5, 8, 1, 0, 0, 0, 4, 4, 0),
-      CoxFeature("Patient_C", 1, 324, "25-29", 0, 3, 0, 1, 0, 2, 0, 0, 0),
-      CoxFeature("Patient_C", 1, 324, "25-29", 3, 5, 1, 1, 0, 3, 0, 0, 0),
-      CoxFeature("Patient_C.1", 1, 324, "25-29", 0, 3, 0, 1, 0, 2, 0, 0, 0),
-      CoxFeature("Patient_C.1", 1, 324, "25-29", 3, 40, 0, 1, 0, 3, 0, 0, 0)
+      CoxFeature("Patient_A", 1, 566, "45-49", 0, 3, 0, 0, 0, 0, 1, 1, 0),
+      CoxFeature("Patient_A", 1, 566, "45-49", 3, 8, 1, 0, 0, 0, 2, 2, 0),
+      CoxFeature("Patient_C", 1, 324, "25-29", 0, 3, 0, 1, 0, 1, 0, 0, 0),
+      CoxFeature("Patient_C", 1, 324, "25-29", 3, 5, 1, 1, 0, 2, 0, 0, 0),
+      CoxFeature("Patient_C.1", 1, 324, "25-29", 0, 3, 0, 1, 0, 1, 0, 0, 0),
+      CoxFeature("Patient_C.1", 1, 324, "25-29", 3, 40, 0, 1, 0, 2, 0, 0, 0)
     ).toDF
 
     // When
diff --git a/src/test/scala/fr/polytechnique/cmap/cnam/filtering/exposures/DosageBasedWeightAggSuite.scala b/src/test/scala/fr/polytechnique/cmap/cnam/filtering/exposures/DosageBasedWeightAggSuite.scala
@@ -35,16 +35,16 @@ class DosageBasedWeightAggSuite extends SharedContext{
     ).toDF("PatientID", "category", "eventId", "start", "exposureStart", "weight")
 
     val expected = Seq(
-      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 1, 1)), Some(1)),
-      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 1, 1)), Some(1)),
-      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 8, 1)), Some(2)),
-      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 10, 1)), Some(4)),
-      ("Patient_A", "SULFONYLUREA", Some(makeTS(2008, 2, 1)), Some(1)),
-      ("Patient_A", "SULFONYLUREA", Some(makeTS(2008, 7, 5)), Some(2)),
-      ("Patient_A", "SULFONYLUREA", Some(makeTS(2008, 8, 1)), Some(3)),
-      ("Patient_B", "SULFONYLUREA", Some(makeTS(2008, 2, 1)), Some(2)),
-      ("Patient_B", "SULFONYLUREA", Some(makeTS(2008, 2, 1)), Some(2)),
-      ("Patient_B", "SULFONYLUREA", Some(makeTS(2008, 11, 1)), Some(4))
+      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 1, 1)), Some(1.0)),
+      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 1, 1)), Some(1.0)),
+      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 8, 1)), Some(2.0)),
+      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 10, 1)), Some(4.0)),
+      ("Patient_A", "SULFONYLUREA", Some(makeTS(2008, 2, 1)), Some(1.0)),
+      ("Patient_A", "SULFONYLUREA", Some(makeTS(2008, 7, 5)), Some(2.0)),
+      ("Patient_A", "SULFONYLUREA", Some(makeTS(2008, 8, 1)), Some(3.0)),
+      ("Patient_B", "SULFONYLUREA", Some(makeTS(2008, 2, 1)), Some(2.0)),
+      ("Patient_B", "SULFONYLUREA", Some(makeTS(2008, 2, 1)), Some(2.0)),
+      ("Patient_B", "SULFONYLUREA", Some(makeTS(2008, 11, 1)), Some(4.0))
     ).toDF("PatientID", "eventId", "exposureStart", "weight")
 
     // When
@@ -90,16 +90,16 @@ class DosageBasedWeightAggSuite extends SharedContext{
     ).toDF("PatientID", "category", "eventId", "start", "exposureStart","weight")
 
     val expected = Seq(
-      ("Patient_A", "PIOGLITAZONE", makeTS(2008, 1, 1), Some(1)),
-      ("Patient_A", "PIOGLITAZONE", makeTS(2008, 1, 1), Some(1)),
-      ("Patient_A", "PIOGLITAZONE", makeTS(2008, 1, 1), Some(1)),
-      ("Patient_A", "PIOGLITAZONE", makeTS(2008, 1, 1), Some(1)),
-      ("Patient_A", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1)),
-      ("Patient_A", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1)),
-      ("Patient_A", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1)),
-      ("Patient_B", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1)),
-      ("Patient_B", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1)),
-      ("Patient_B", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1))
+      ("Patient_A", "PIOGLITAZONE", makeTS(2008, 1, 1), Some(1.0)),
+      ("Patient_A", "PIOGLITAZONE", makeTS(2008, 1, 1), Some(1.0)),
+      ("Patient_A", "PIOGLITAZONE", makeTS(2008, 1, 1), Some(1.0)),
+      ("Patient_A", "PIOGLITAZONE", makeTS(2008, 1, 1), Some(1.0)),
+      ("Patient_A", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1.0)),
+      ("Patient_A", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1.0)),
+      ("Patient_A", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1.0)),
+      ("Patient_B", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1.0)),
+      ("Patient_B", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1.0)),
+      ("Patient_B", "SULFONYLUREA", makeTS(2008, 2, 1), Some(1.0))
     ).toDF("PatientID", "eventId", "exposureStart", "weight")
 
     // When
@@ -141,10 +141,10 @@ class DosageBasedWeightAggSuite extends SharedContext{
     ).toDF("PatientID", "category", "eventId", "start", "exposureStart","weight")
 
     val expected = Seq(
-      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 3, 1)), Some(1)),
-      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 8, 1)), Some(3)),
-      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 8, 1)), Some(3)),
-      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 10, 1)), Some(4))
+      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 3, 1)), Some(1.0)),
+      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 8, 1)), Some(3.0)),
+      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 8, 1)), Some(3.0)),
+      ("Patient_A", "PIOGLITAZONE", Some(makeTS(2008, 10, 1)), Some(4.0))
     ).toDF("PatientID", "eventId", "exposureStart", "weight")
 
     // When
diff --git a/src/test/scala/fr/polytechnique/cmap/cnam/filtering/exposures/ExposuresTransformerSuite.scala b/src/test/scala/fr/polytechnique/cmap/cnam/filtering/exposures/ExposuresTransformerSuite.scala
diff --git a/src/test/scala/fr/polytechnique/cmap/cnam/filtering/exposures/PurchaseBasedWeightAggSuite.scala b/src/test/scala/fr/polytechnique/cmap/cnam/filtering/exposures/PurchaseBasedWeightAggSuite.scala

Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,8 @@ default = {`
`57`	`57`	`window = 1 #Number of months to quantile.`
`58`	`58`	`start_threshold = 6 #Number of months within which more than one purchases have to made`
`59`	`59`	`end_threshold = 4 #Number of months during which no purchases of the particular molecule have to be made`
`60`		`- dosage_level_intervals = [0,100,200,300,400,500] #list of consumption levels in mg / put only 0 when we want all the weights to 1`
	`60`	`+ dosage_level_intervals = [0, 100, 200, 300, 400, 500] #list of consumption levels in mg / put only 0 when we want all the weights to 1`
	`61`	`+ purchase_intervals = [0, 3, 5] #list of consumption levels in mg / put only 0 when we want all the weights to 1`
`61`	`62`	`}`
`62`	`63`	`}`
`63`	`64`
Original file line number	Diff line number	Diff line change
`@@ -154,6 +154,7 @@ object FilteringConfig {`
`154`	`154`	`cumulativeExposureWindow = conf.getInt("exposures.cumulative.window"),`
`155`	`155`	`cumulativeStartThreshold = conf.getInt("exposures.cumulative.start_threshold"),`
`156`	`156`	`cumulativeEndThreshold = conf.getInt("exposures.cumulative.end_threshold"),`
`157`		`- dosageLevelIntervals = conf.getIntList("exposures.cumulative.dosage_level_intervals").asScala.toList.map(x => x.toInt)`
	`157`	`+ dosageLevelIntervals = conf.getIntList("exposures.cumulative.dosage_level_intervals").asScala.map(_.toInt).toList,`
	`158`	`+ purchaseIntervals = conf.getIntList("exposures.cumulative.purchase_intervals").asScala.map(_.toInt).toList`
`158`	`159`	`)`
`159`	`160`	`}`
Original file line number	Diff line number	Diff line change
`@@ -12,7 +12,8 @@ class TimeBasedWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data) {`
`12`	`12`	`cumWindow: Option[Int],`
`13`	`13`	`cumStartThreshold: Option[Int],`
`14`	`14`	`cumEndThreshold: Option[Int],`
`15`		`- dosageLevelIntervals: Option[List[Int]]): DataFrame = {`
	`15`	`+ dosageLevelIntervals: Option[List[Int]],`
	`16`	`+ purchaseIntervals: Option[List[Int]]): DataFrame = {`
`16`	`17`
`17`	`18`	`val window = Window.partitionBy("patientID", "eventId").orderBy("exposureStart", "exposureEnd")`
`18`	`19`	`data`
`@@ -22,5 +23,5 @@ class TimeBasedWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data) {`
`22`	`23`	`.withColumn("exposureEnd", col("followUpEnd"))`
`23`	`24`	`}`
`24`	`25`
`25`		`- def aggregateWeight: DataFrame = aggregateWeight(None, None, None, None, None)`
	`26`	`+ def aggregateWeight: DataFrame = aggregateWeight(None, None, None, None, None, None)`
`26`	`27`	`}`