<commit_before>"""Computes and updates the `direction` column in the `covidcast` table.

Update is only performed for stale rows.
"""

# standard library
import argparse

# third party
import numpy as np

# first party
from delphi.epidata.acquisition.covidcast.database import Database
from delphi.epidata.acquisition.covidcast.direction import Direction


class Constants:
  """Constants used for direction classification."""

  # Display height is currently defined as 6 standard deviations of the
  # historical values of a particular signal across all locations at a
  # particular geographic resolution.
  SIGNAL_STDEV_SCALE = 6

  # Historical data is roughly defined to be all signal values up to, and
  # including, but not after, 2020-04-22.
  SIGNAL_STDEV_MAX_DAY = 20200422

  # Direction is non-zero when the following two conditions are met, given a
  # regression fit on a 7-day trend:
  #   - slope is statistically significant (±1 standard error)
  #   - slope is subjectively up or down (±10% of display height, per day)
  TREND_NUM_DAYS = 7
  SLOPE_STERR_SCALE = 1
  SLOPE_PERCENT_CHANGE = 0.1
  BASE_SLOPE_THRESHOLD = (
      SIGNAL_STDEV_SCALE * (SLOPE_PERCENT_CHANGE / TREND_NUM_DAYS)
  )


def get_argument_parser():
  """Define command line arguments."""

  # there are no flags, but --help will still work
  return argparse.ArgumentParser()


def update_loop(database, direction_impl=Direction):
  """Find and update rows with a stale `direction` value.

  `database`: an open connection to the epidata database
  """

  # get the keys for time-seres which may be stale
  stale_series = database.get_keys_with_potentially_stale_direction()
  num_series = len(stale_series)
  num_rows = 0
  for row in stale_series:
    num_rows += row[-1]
  msg = 'found %d time-series (%d rows) which may have stale direction'
  print(msg % (num_series, num_rows))

  # get the historical standard deviation of all signals and resolutions
  rows = database.get_data_stdev_across_locations(
      Constants.SIGNAL_STDEV_MAX_DAY)
  data_stdevs = {}
  for (source, signal, geo_type, aggregate_stdev) in rows:
    if source not in data_stdevs:
      data_stdevs[source] = {}
    if signal not in data_stdevs[source]:
      data_stdevs[source][signal] = {}
    data_stdevs[source][signal][geo_type] = aggregate_stdev

  # update direction for each time-series
  for ts_index, row in enumerate(stale_series):
    (
      source,
      signal,
      geo_type,
      geo_value,
      max_timestamp1,
      min_timestamp2,
      min_day,
      max_day,
      series_length,
    ) = row

    # progress reporting for anyone debugging/watching the output
    be_verbose = ts_index < 100<commit_msg>Skip computing direction on signals without enough history<commit_after>"""Computes and updates the `direction` column in the `covidcast` table.

Update is only performed for stale rows.
"""

# standard library
import argparse

# third party
import numpy as np

# first party
from delphi.epidata.acquisition.covidcast.database import Database
from delphi.epidata.acquisition.covidcast.direction import Direction


class Constants:
  """Constants used for direction classification."""

  # Display height is currently defined as 6 standard deviations of the
  # historical values of a particular signal across all locations at a
  # particular geographic resolution.
  SIGNAL_STDEV_SCALE = 6

  # Historical data is roughly defined to be all signal values up to, and
  # including, but not after, 2020-04-22.
  SIGNAL_STDEV_MAX_DAY = 20200422

  # Direction is non-zero when the following two conditions are met, given a
  # regression fit on a 7-day trend:
  #   - slope is statistically significant (±1 standard error)
  #   - slope is subjectively up or down (±10% of display height, per day)
  TREND_NUM_DAYS = 7
  SLOPE_STERR_SCALE = 1
  SLOPE_PERCENT_CHANGE = 0.1
  BASE_SLOPE_THRESHOLD = (
      SIGNAL_STDEV_SCALE * (SLOPE_PERCENT_CHANGE / TREND_NUM_DAYS)
  )


def get_argument_parser():
  """Define command line arguments."""

  # there are no flags, but --help will still work
  return argparse.ArgumentParser()


def update_loop(database, direction_impl=Direction):
  """Find and update rows with a stale `direction` value.

  `database`: an open connection to the epidata database
  """

  # get the keys for time-seres which may be stale
  stale_series = database.get_keys_with_potentially_stale_direction()
  num_series = len(stale_series)
  num_rows = 0
  for row in stale_series:
    num_rows += row[-1]
  msg = 'found %d time-series (%d rows) which may have stale direction'
  print(msg % (num_series, num_rows))

  # get the historical standard deviation of all signals and resolutions
  rows = database.get_data_stdev_across_locations(
      Constants.SIGNAL_STDEV_MAX_DAY)
  data_stdevs = {}
  for (source, signal, geo_type, aggregate_stdev) in rows:
    if source not in data_stdevs:
      data_stdevs[source] = {}
    if signal not in data_stdevs[source]:
      data_stdevs[source][signal] = {}
    data_stdevs[source][signal][geo_type] = aggregate_stdev

  # update direction for each time-series
  for ts_index, row in enumerate(stale_series):
    (
      source,
      signal,
      geo_type,
      geo_value,
      max_timestamp1,
      min_timestamp2,
      min_day,
      max_day,
      series_length,
    ) = row

    if source not in data_stdevs or signal not in data_stdevs[source]:
      msg = '%s.%s is too new a signal to have a historical stdev reference; skipping'
      print(msg % (source,signal))
      continue

    # progress reporting for anyone debugging/watching the output
    be_verbose = ts_index < 100
{'fraction_non_alphanumeric': 0.07406094968107725, 'fraction_numerical': 0.011693834160170093, 'mean_word_length': 3.755686604886268, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5099299', 'n_tokens_mistral': 1843, 'n_tokens_neox': 1699, 'n_words': 583}