Dataframe Time Series Resampling and rolling calculations like python pandas dataframe

[Kanish2002]

pandas.DataFrame.resample
pandas.DataFrame.rolling

How to do resampling time series and rolling calculations using Kotlin Dataframe.
Is there any development going on for that or available third party libraries?

[Jolanrensen]

Relevant issue #95

[Kanish2002]

Relevant issue #95

Yeah it's #95 is fine for rolling calculations, but sir, I want to resample DataColumn<java.sql.Timestamp>,DataColumn based on frequencies.
pandas.DataFrame.resample like this function.

https://blog.quantinsti.com/tick-tick-ohlc-data-pandas-tutorial/
Check the above link, I need the exact output like that, to handle data in a generic way.

I'm working on a time series data, which maybe of any frequency and I want to convert it into a specific frequency.

  1 2023-03-20 09:15:00.0    1801.15
  2 2023-03-20 09:15:02.0    1801.15
  3 2023-03-20 09:15:03.0    1801.15
  4 2023-03-20 09:15:03.0    1801.15
  5 2023-03-20 09:15:05.0    1801.15
  6 2023-03-20 09:15:06.0    1801.15
  7 2023-03-20 09:15:06.0    1801.15
  8 2023-03-20 09:15:08.0    1801.15
  9 2023-03-20 09:15:09.0    1801.15
 10 2023-03-20 09:15:11.0    1801.15
 11 2023-03-20 09:15:12.0    1801.15
 12 2023-03-20 09:15:12.0    1801.15
 13 2023-03-20 09:15:13.0    1801.15
 14 2023-03-20 09:15:15.0    1801.15
 15 2023-03-20 09:15:16.0    1801.15
 16 2023-03-20 09:15:17.0    1801.15
 17 2023-03-20 09:15:18.0    1801.15
 18 2023-03-20 09:15:18.0    1801.15
 19 2023-03-20 09:15:19.0    1801.15

convert a dataframe like the above into say for example '30Seconds' Frequency..

                          open          high             low        close       mean
1 2023-03-20 09:15:00.0    1801.15     1801.15         1801.15    1801.15     1801.15
2 2023-03-20 09:15:30.0    1801.15     1801.15         1801.15    1801.15     1801.15
1 2023-03-20 09:16:00.0    1801.15     1801.15         1801.15    1801.15     1801.15
2 2023-03-20 09:16:30.0    1801.15     1801.15         1801.15    1801.15     1801.15

[Jolanrensen]

I see, could you share an example of your data? We don't have a resample function currently, but we might be able to reproduce the same thing in Kotlin/DataFrame :)

[Kanish2002]

I see, could you share an example of your data? We don't have a resample function currently, but we might be able to reproduce the same thing in Kotlin/DataFrame :)

I think you generate some sample data for the above problem,
but anyway I'm sharing some code to generate some samples

import org.jetbrains.kotlinx.dataframe.api.*
import java.sql.Timestamp
import java.util.*
import kotlin.random.Random
fun main() {
    val cal = Calendar.getInstance()
    cal.set(2023, 0, 1, 9, 15, 0)

    val nrows = 10000
    val date_df = (1..nrows).map {
        Timestamp(cal.time.time)
            .also { cal.add(Random.nextInt(13, 15), Random.nextInt(0, 3)) }
    }.toColumn("Timestamp")
    val price = (1..nrows).map { Random.nextInt(50, 1000) }.toColumn("price")

    val dataFrame = dataFrameOf(date_df,price)
    println(dataFrame)
}

The above code generates sample dataframe of which contains DataColumn<java.sql.Timestamp> and Int you can generate sample data by using it.

https://pandas.pydata.org/docs/reference/api/pandas.core.resample.Resampler.ohlc.html#
I need a TimeGrouper with some aggregations over it, I think aggregations are easy, but TimeGrouper based on a specified frequency is a little bit tricky one.

https://pandas.pydata.org/docs/reference/api/pandas.core.resample.Resampler.bfill.html#
https://pandas.pydata.org/docs/reference/api/pandas.core.resample.Resampler.ffill.html#

and Also bfill() and ffill() on Datacolumns

[koperagen]

Try this on 0.10.0-dev-1515 to use toStandaloneHTML function.
I think it looks somewhat like what you want, doesn't it? Should work as a workaround for now, or at least give you a starting point. Grouping time frame can be adjusted with val frame = 30000. Then, after groupBy aggregation should be possible

import org.jetbrains.kotlinx.dataframe.api.*
import java.sql.Timestamp
import java.util.*
import kotlin.random.Random
import org.jetbrains.kotlinx.dataframe.io.DisplayConfiguration
import org.jetbrains.kotlinx.dataframe.io.toStandaloneHTML

fun main() {
    val cal = Calendar.getInstance()
    cal.set(2023, 0, 1, 9, 15, 0)

    val nrows = 10000
    val date_df = (1..nrows).map {
        Timestamp(cal.time.time)
            .also { cal.add(Random.nextInt(13, 15), Random.nextInt(0, 3)) }
    }.toColumn("Timestamp")
    val price = (1..nrows).map { Random.nextInt(50, 1000) }.toColumn("price")

    val dataFrame = dataFrameOf(date_df,price)
    println(dataFrame)

//    dataFrame.sortBy(date_df).toStandaloneHTML(DisplayConfiguration(rowsLimit = null)).openInBrowser()

    val frame = 30000

    var accumulated = 0L
    dataFrame
        .sortBy(date_df)
        .add("timestampDiff") {
            diff { date_df().time } ?: 0L
        }
        .cumSum("timestampDiff")
        .add("groupChanged") {
            val value = ("timestampDiff"<Long>() - accumulated) > frame
            if (value) {
                accumulated = "timestampDiff"<Long>()
            }
            value
        }
        .convert("groupChanged").toInt()
        .cumSum("groupChanged")
        .rename("groupChanged").into("groupId")
        .also { it.toStandaloneHTML(DisplayConfiguration(rowsLimit = null)).openInBrowser() }
        .groupBy("groupId")
        .toDataFrame()
        .toStandaloneHTML().openInBrowser()

}

[Jolanrensen]

@koperagen seems like we have been tackling the same thing XD

my version:

import kotlinx.datetime.toJavaInstant
import kotlinx.datetime.toKotlinInstant
import java.time.temporal.ChronoUnit
import java.time.Instant

fun Instant.toNearest30Seconds(): Instant =
    truncatedTo(ChronoUnit.SECONDS).let {
        if (it.epochSecond % 30 == 0L) it
        else it.plusSeconds(30 - it.epochSecond % 30)
    }

dataFrame.groupBy {
    expr("timeKey") { Timestamp.toInstant().toNearest30Seconds() }
}.aggregate {
    price.first() into "open"
    price.max() into "high"
    price.min() into "low"
    price.last() into "close"
    price.mean() into "mean"
}

It's not very reusable, unless we find a way to make the toNearest30Seconds() function generic to make it able to round to any timespan.

[Kanish2002]

@koperagen seems like we have been tackling the same thing XD

my version:

import kotlinx.datetime.toJavaInstant
import kotlinx.datetime.toKotlinInstant
import java.time.temporal.ChronoUnit
import java.time.Instant

fun Instant.toNearest30Seconds(): Instant =
    truncatedTo(ChronoUnit.SECONDS).let {
        if (it.epochSecond % 30 == 0L) it
        else it.plusSeconds(30 - it.epochSecond % 30)
    }

dataFrame.groupBy {
    expr("timeKey") { Timestamp.toInstant().toNearest30Seconds() }
}.aggregate {
    price.first() into "open"
    price.max() into "high"
    price.min() into "low"
    price.last() into "close"
    price.mean() into "mean"
}

It's not very reusable, unless we find a way to make the toNearest30Seconds() function generic to make it able to round to any timespan.

Hi @Jolanrensen and @koperagen Thanks for your replies, actually I want a generic one, where conditon may be of hour based time to seconds or minute based, I may give any type(hourly,seconds,minutes,days) and if I want to resample exactly to seconds(say 15, 20, 45) or minutes(5,15) or any thing like that.

[Kanish2002]

Try this on 0.10.0-dev-1515 to use toStandaloneHTML function. I think it looks somewhat like what you want, doesn't it? Should work as a workaround for now, or at least give you a starting point. Grouping time frame can be adjusted with val frame = 30000. Then, after groupBy aggregation should be possible

import org.jetbrains.kotlinx.dataframe.api.*
import java.sql.Timestamp
import java.util.*
import kotlin.random.Random
import org.jetbrains.kotlinx.dataframe.io.DisplayConfiguration
import org.jetbrains.kotlinx.dataframe.io.toStandaloneHTML

fun main() {
    val cal = Calendar.getInstance()
    cal.set(2023, 0, 1, 9, 15, 0)

    val nrows = 10000
    val date_df = (1..nrows).map {
        Timestamp(cal.time.time)
            .also { cal.add(Random.nextInt(13, 15), Random.nextInt(0, 3)) }
    }.toColumn("Timestamp")
    val price = (1..nrows).map { Random.nextInt(50, 1000) }.toColumn("price")

    val dataFrame = dataFrameOf(date_df,price)
    println(dataFrame)

//    dataFrame.sortBy(date_df).toStandaloneHTML(DisplayConfiguration(rowsLimit = null)).openInBrowser()

    val frame = 30000

    var accumulated = 0L
    dataFrame
        .sortBy(date_df)
        .add("timestampDiff") {
            diff { date_df().time } ?: 0L
        }
        .cumSum("timestampDiff")
        .add("groupChanged") {
            val value = ("timestampDiff"<Long>() - accumulated) > frame
            if (value) {
                accumulated = "timestampDiff"<Long>()
            }
            value
        }
        .convert("groupChanged").toInt()
        .cumSum("groupChanged")
        .rename("groupChanged").into("groupId")
        .also { it.toStandaloneHTML(DisplayConfiguration(rowsLimit = null)).openInBrowser() }
        .groupBy("groupId")
        .toDataFrame()
        .toStandaloneHTML().openInBrowser()

}

hi @koperagen can I use frame size and resample data with frequency correct ?
for example if I want to resample for 2 minutes I can set frame size as 120000 and resample it ?

[Jolanrensen]

import kotlinx.datetime.Instant
import kotlinx.datetime.toJavaInstant
import kotlinx.datetime.toKotlinInstant
import kotlin.time.Duration.Companion.seconds
import kotlin.time.Duration.Companion.days
import kotlin.time.Duration.Companion.minutes // etc
import kotlin.time.Duration

fun Instant.roundDownTo(duration: Duration): Instant {
    val durationMillis = duration.inWholeMilliseconds
    val epochMillis = toEpochMilliseconds()
    return if (epochMillis % durationMillis == 0L)
        this
    else
        Instant.fromEpochMilliseconds(epochMillis - epochMillis % durationMillis)
}

dataFrame.groupBy {
    expr("timeKey") { Timestamp.toInstant().toKotlinInstant().roundDownTo(10.seconds) }
}.aggregate {
    price.first() into "open"
    price.max() into "high"
    price.min() into "low"
    price.last() into "close"
    price.mean() into "mean"
}

Made a generic version using Kotlin Instant and Duration. It's not exactly resampling, just grouping the time values, floored by the closest multiple of duration

[Kanish2002]

import kotlinx.datetime.Instant
import kotlinx.datetime.toJavaInstant
import kotlinx.datetime.toKotlinInstant
import kotlin.time.Duration.Companion.seconds
import kotlin.time.Duration.Companion.days
import kotlin.time.Duration.Companion.minutes // etc
import kotlin.time.Duration

fun Instant.roundDownTo(duration: Duration): Instant {
    val durationMillis = duration.inWholeMilliseconds
    val epochMillis = toEpochMilliseconds()
    return if (epochMillis % durationMillis == 0L)
        this
    else
        Instant.fromEpochMilliseconds(epochMillis - epochMillis % durationMillis)
}

dataFrame.groupBy {
    expr("timeKey") { Timestamp.toInstant().toKotlinInstant().roundDownTo(10.seconds) }
}.aggregate {
    price.first() into "open"
    price.max() into "high"
    price.min() into "low"
    price.last() into "close"
    price.mean() into "mean"
}

Made a generic version using Kotlin Instant and Duration. It's not exactly resampling, just grouping the time values, floored by the closest multiple of duration

Can Hourly Data converted to minute or seconds , like kind of backfill and forward fill timestamp and price?

[Kanish2002]

I'm glad with your solution @koperagen and @Jolanrensen, my exact problem is to resample any time series data into a particular frequency. It maybe of Upsampling or Downsampling .
Need a resampling method exactly like pandas.DataFrame.resample and bfill and ffill.

Like If the timestamp is missing, it should added in a newly sampled data.
Kindly check pandas.DataFrame.resample you will get better idea about my issue

[Jolanrensen]

DataFrame does not support a direct resample as of now. But you can downsample using something we described and upsample by creating your own backfill and frontfill method.
If you are building a solution yourself, feel free to add it to the library :)

[koperagen]

hi @koperagen can I use frame size and resample data with frequency correct ?
for example if I want to resample for 2 minutes I can set frame size as 120000 and resample it ?

Yes, that should work. Following code produces a bit different result from @Jolanrensen as it uses a rolling sum to calculate windows, but again, that should just give you an idea for workaround until this API is introduced (not in the near future)

fun main() {
    val cal = Calendar.getInstance()
    cal.set(2023, 0, 1, 9, 15, 0)

    val nrows = 10000
    val date_df = (1..nrows).map {
        Timestamp(cal.time.time)
            .also { cal.add(Random.nextInt(13, 15), Random.nextInt(0, 3)) }
    }.toColumn("Timestamp")
    val price = (1..nrows).map { Random.nextInt(50, 1000) }.toColumn("price")

    val dataFrame = dataFrameOf(date_df,price)
    println(dataFrame)
    val frame = 30.seconds
    
    dataFrame.resample(frame) { date_df().time }.aggregate {
        price().first() into "open"
        price().max() into "high"
        price().min() into "low"
        price().last() into "close"
        price().mean() into "mean"
    }.toStandaloneHTML().openInBrowser()
}

fun AnyFrame.resample(
    frame: Duration,
    expression: RowExpression<*, Long>
): GroupBy<*, *> {
    var accumulated = 0L
    return sortBy { expr("", expression = expression) }
        .add("timestampDiff") {
            diff(expression) ?: 0L
        }
        .cumSum("timestampDiff")
        .add("groupChanged") {
            val value = ("timestampDiff"<Long>() - accumulated) > frame.inWholeMilliseconds
            if (value) {
                accumulated = "timestampDiff"<Long>()
            }
            value
        }
        .convert("groupChanged").toInt()
        .cumSum("groupChanged")
        .rename("groupChanged").into("groupId")
        .groupBy("groupId")
}

[Kanish2002]

@koperagen seems like we have been tackling the same thing XD

my version:

import kotlinx.datetime.toJavaInstant
import kotlinx.datetime.toKotlinInstant
import java.time.temporal.ChronoUnit
import java.time.Instant

fun Instant.toNearest30Seconds(): Instant =
    truncatedTo(ChronoUnit.SECONDS).let {
        if (it.epochSecond % 30 == 0L) it
        else it.plusSeconds(30 - it.epochSecond % 30)
    }

dataFrame.groupBy {
    expr("timeKey") { Timestamp.toInstant().toNearest30Seconds() }
}.aggregate {
    price.first() into "open"
    price.max() into "high"
    price.min() into "low"
    price.last() into "close"
    price.mean() into "mean"
}

It's not very reusable, unless we find a way to make the toNearest30Seconds() function generic to make it able to round to any timespan.

import kotlinx.datetime.Instant
import kotlinx.datetime.toJavaInstant
import kotlinx.datetime.toKotlinInstant
import kotlin.time.Duration.Companion.seconds
import kotlin.time.Duration.Companion.days
import kotlin.time.Duration.Companion.minutes // etc
import kotlin.time.Duration

fun Instant.roundDownTo(duration: Duration): Instant {
    val durationMillis = duration.inWholeMilliseconds
    val epochMillis = toEpochMilliseconds()
    return if (epochMillis % durationMillis == 0L)
        this
    else
        Instant.fromEpochMilliseconds(epochMillis - epochMillis % durationMillis)
}

dataFrame.groupBy {
    expr("timeKey") { Timestamp.toInstant().toKotlinInstant().roundDownTo(10.seconds) }
}.aggregate {
    price.first() into "open"
    price.max() into "high"
    price.min() into "low"
    price.last() into "close"
    price.mean() into "mean"
}

Made a generic version using Kotlin Instant and Duration. It's not exactly resampling, just grouping the time values, floored by the closest multiple of duration

I think you have written it for datalore/jupyter can you rewrite that for intellijIDEA string accessor

[Jolanrensen]

Let that be an exercise for you