markcutajar

"""
This code attempts to reproduce:
Adaptive Normalization: A novel data normalization approach for non-stationary time series

Conference: International Joint Conference on Neural Networks, IJCNN 2010, Barcelona, Spain, 18-23 July, 2010
Eduardo Ogasawara et Al.

"""

markcutajar

Google Analytics 4 in Hugo

Blog post with slightly more detail can be found here. If you want to just get to it, read on.

I (very) recently decided to reimplement GA in my site, and found that existing implementation in Hugo was not compatible with GA4's new Measurement ID. This is an easy way to drop your Measurement ID into your site. I'm not going to go into how to sign up for GA.

This implementation requires that you create a site parameter for analytics, create a partial, and call the partial. I tweaked the names of the parameters and files so they didn't collide with the built-in hugo code.

Place the GoogleAnalyicsID (Measurement ID) in config.toml within [params].

markcutajar

import logging
import argparse

from fxbeam.fxbeam import FxBeam


if __name__ == '__main__':

    logging.getLogger().setLevel(logging.INFO)
    parser = argparse.ArgumentParser()

markcutajar

class CustomTestCase(unittest.TestCase):
    def setUp(self):
        pipeline_options = PipelineOptions(['--runner=DirectRunner'])
        pipeline_options.view_as(SetupOptions).save_main_session = True
        self.pipeline = beam.Pipeline(options=pipeline_options)


class TestFunctionX(CustomTestCase):
    def test_run(self):
        with self.pipeline:

markcutajar

@staticmethod
def get_value(item):
    """Function to calculate the value of the element.
    This is done so we can easily change how the value is calculated,
    by changing it in this one place.
    :param item: Tick item
    :return: Value of tick item
    """
    return item['ask']

markcutajar

class CustomCombineFn(beam.CombineFn):
  def create_accumulator(self):
    return # create accumulator here

  def add_input(self, current_accumulation, input):
    return # perform accumulation here

  def merge_accumulators(self, accumulators):
    return # merge accumulators running un different nodes

markcutajar

def map_elements(self, data):
      """Map function to create key:value pairs to run CombinePerKey function.
      :param data: PCollection being processed with time_group_key column and
          instrument_column if set to use.
      :return: PCollection with mapped data
      """
      action_name = 'Resampler - Map data'
      if self.instrument_column:
          return data | action_name >> beam.Map(
              lambda x: ((x['time_group_key'], x[self.instrument_column]), x)

markcutajar

...
yield {
    **element, 
    time_group_key: element['timestamp'] // window_size
}

markcutajar

class AddTimestamp(beam.DoFn):
    """ParDo take a timestamp column and assign it 
		to row to produce a timestamped value
		"""
    def process(self, element, timestamp_key, delete_key=False, **kwargs):
        """
        :param element: Element being processed
        :param timestamp_key: Field containing the timestamp
        :param delete_key: Whether to delete the original timestamp or not
        :return: Elements with a datetime filed

markcutajar

rows = rows | 'Convert to timestamp field' >> beam.ParDo(
    ToTimestamp(),
    string_format=self.TICK_DATA_TIMESTAMP_FORMAT,
    datetime_key=self.input_columns[0],
    timestamp_key='timestamp'
)

rows = rows | 'Convert to datetime object' >> beam.ParDo(
    AddTimestamp(),
    timestamp_key='timestamp'