Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .bumpversion.cfg
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
[bumpversion]
current_version = 0.2.2
current_version = 0.2.3
commit = True
message = chore: bump covidcast-indicators to {new_version}
tag = False
2 changes: 1 addition & 1 deletion _delphi_utils_python/.bumpversion.cfg
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
[bumpversion]
current_version = 0.2.2
current_version = 0.2.3
commit = True
message = chore: bump delphi_utils to {new_version}
tag = False
Expand Down
6 changes: 3 additions & 3 deletions _delphi_utils_python/.pylintrc
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,9 @@ disable=logging-format-interpolation,
[BASIC]

# Allow arbitrarily short-named variables.
variable-rgx=[a-z_][a-z0-9_]*
argument-rgx=[a-z_][a-z0-9_]*
attr-rgx=[a-z_][a-z0-9_]*
variable-rgx=([a-z_][a-z0-9_]*|[a-zA-Z])
argument-rgx=([a-z_][a-z0-9_]*|[a-zA-Z])
attr-rgx=([a-z_][a-z0-9_]*|[a-zA-Z])

[DESIGN]

Expand Down
3 changes: 2 additions & 1 deletion _delphi_utils_python/delphi_utils/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,5 +13,6 @@
from .smooth import Smoother
from .signal import add_prefix
from .nancodes import Nans
from .weekday import Weekday

__version__ = "0.2.2"
__version__ = "0.2.3"
Original file line number Diff line number Diff line change
Expand Up @@ -3,23 +3,48 @@

Created: 2020-05-06
"""



# third party
import cvxpy as cp
from cvxpy.error import SolverError
import numpy as np

# first party
from .config import Config
from cvxpy.error import SolverError


class Weekday:
"""Class to handle weekday effects."""

@staticmethod
def get_params(data, logger):
def get_params(data, denominator_col, numerator_cols, date_col, scales, logger):
r"""Fit weekday correction for each col in numerator_cols.

Return a matrix of parameters: the entire vector of betas, for each time
series column in the data.
"""
tmp = data.reset_index()
denoms = tmp.groupby(date_col).sum()[denominator_col]
nums = tmp.groupby(date_col).sum()[numerator_cols]

# Construct design matrix to have weekday indicator columns and then day
# indicators.
X = np.zeros((nums.shape[0], 6 + nums.shape[0]))
not_sunday = np.where(nums.index.dayofweek != 6)[0]
X[not_sunday, np.array(nums.index.dayofweek)[not_sunday]] = 1
X[np.where(nums.index.dayofweek == 6)[0], :6] = -1
X[:, 6:] = np.eye(X.shape[0])

npnums, npdenoms = np.array(nums), np.array(denoms)
params = np.zeros((nums.shape[1], X.shape[1]))

# Loop over the available numerator columns and smooth each separately.
for i in range(nums.shape[1]):
result = Weekday._fit(X, scales, npnums[:, i], npdenoms)
if result is None:
logger.error("Unable to calculate weekday correction")
else:
params[i,:] = result

return params

@staticmethod
def _fit(X, scales, npnums, npdenoms):
r"""Correct a signal estimated as numerator/denominator for weekday effects.

The ordinary estimate would be numerator_t/denominator_t for each time point
Expand Down Expand Up @@ -53,57 +78,31 @@ def get_params(data, logger):

ll = (numerator * (X*b + log(denominator)) - sum(exp(X*b) + log(denominator)))
/ num_days

Return a matrix of parameters: the entire vector of betas, for each time
series column in the data.
"""
denoms = data.groupby(Config.DATE_COL).sum()["Denominator"]
nums = data.groupby(Config.DATE_COL).sum()[Config.CLI_COLS + Config.FLU1_COL]

# Construct design matrix to have weekday indicator columns and then day
# indicators.
X = np.zeros((nums.shape[0], 6 + nums.shape[0]))
not_sunday = np.where(nums.index.dayofweek != 6)[0]
X[not_sunday, np.array(nums.index.dayofweek)[not_sunday]] = 1
X[np.where(nums.index.dayofweek == 6)[0], :6] = -1
X[:, 6:] = np.eye(X.shape[0])

npnums, npdenoms = np.array(nums), np.array(denoms)
params = np.zeros((nums.shape[1], X.shape[1]))

# Loop over the available numerator columns and smooth each separately.
for i in range(nums.shape[1]):
b = cp.Variable((X.shape[1]))

lmbda = cp.Parameter(nonneg=True)
lmbda.value = 10 # Hard-coded for now, seems robust to changes

ll = (
cp.matmul(npnums[:, i], cp.matmul(X, b) + np.log(npdenoms))
- cp.sum(cp.exp(cp.matmul(X, b) + np.log(npdenoms)))
) / X.shape[0]
penalty = (
lmbda * cp.norm(cp.diff(b[6:], 3), 1) / (X.shape[0] - 2)
) # L-1 Norm of third differences, rewards smoothness
scales = [1, 1e5, 1e10, 1e15]
for scale in scales:
try:
prob = cp.Problem(cp.Minimize((-ll + lmbda * penalty) / scale))
_ = prob.solve()
params[i,:] = b.value
break
except SolverError:
# If the magnitude of the objective function is too large, an error is
# thrown; Rescale the objective function by going through loop
pass
else:
# Leaving params[i,:] = 0 is equivalent to not performing weekday correction
logger.error("Unable to calculate weekday correction")

return params
b = cp.Variable((X.shape[1]))

lmbda = cp.Parameter(nonneg=True)
lmbda.value = 10 # Hard-coded for now, seems robust to changes

ll = (
cp.matmul(npnums, cp.matmul(X, b) + np.log(npdenoms)) -
cp.sum(cp.exp(cp.matmul(X, b) + np.log(npdenoms)))
) / X.shape[0]
# L-1 Norm of third differences, rewards smoothness
penalty = lmbda * cp.norm(cp.diff(b[6:], 3), 1) / (X.shape[0] - 2)
for scale in scales:
try:
prob = cp.Problem(cp.Minimize((-ll + lmbda * penalty) / scale))
_ = prob.solve()
return b.value
except SolverError:
# If the magnitude of the objective function is too large, an error is
# thrown; Rescale the objective function by going through loop
continue
return None

@staticmethod
def calc_adjustment(params, sub_data):
def calc_adjustment(params, sub_data, cols, date_col):
"""Apply the weekday adjustment to a specific time series.

Extracts the weekday fixed effects from the parameters and uses these to
Expand All @@ -122,14 +121,15 @@ def calc_adjustment(params, sub_data):
-- this has the same effect.

"""
for i, c in enumerate(Config.CLI_COLS + Config.FLU1_COL):
wd_correction = np.zeros((len(sub_data[c])))
tmp = sub_data.copy()

for i, c in enumerate(cols):
wd_correction = np.zeros((len(tmp[c])))

for wd in range(7):
mask = sub_data[Config.DATE_COL].dt.dayofweek == wd
wd_correction[mask] = sub_data.loc[mask, c] / (
mask = tmp[date_col].dt.dayofweek == wd
wd_correction[mask] = tmp.loc[mask, c] / (
np.exp(params[i, wd]) if wd < 6 else np.exp(-np.sum(params[i, :6]))
)
sub_data.loc[:, c] = wd_correction

return sub_data
tmp.loc[:, c] = wd_correction
return tmp
3 changes: 2 additions & 1 deletion _delphi_utils_python/setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
required = [
"boto3",
"covidcast",
"cvxpy",
"epiweeks",
"freezegun",
"gitpython",
Expand All @@ -25,7 +26,7 @@

setup(
name="delphi_utils",
version="0.2.2",
version="0.2.3",
description="Shared Utility Functions for Indicators",
long_description=long_description,
long_description_content_type="text/markdown",
Expand Down
74 changes: 74 additions & 0 deletions _delphi_utils_python/tests/test_weekday.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
import logging

import numpy as np
import pandas as pd
from delphi_utils.weekday import Weekday


class TestWeekday:

TEST_DATA = pd.DataFrame({
"num": np.arange(1, 11, 1),
"den": np.arange(11, 21, 1),
"date": pd.date_range("2020-01-01", "2020-01-10")
})

def test_get_params(self):
TEST_LOGGER = logging.getLogger()

result = Weekday.get_params(self.TEST_DATA, "den", ["num"], "date", [1], TEST_LOGGER)
print(result)
expected_result = [
-0.05993665,
-0.0727396,
-0.05618517,
0.0343405,
0.12534997,
0.04561813,
-2.27669028,
-1.89564374,
-1.5695407,
-1.29838116,
-1.08216513,
-0.92089259,
-0.81456355,
-0.76317802,
-0.76673598,
-0.82523745,
]
assert np.allclose(result, expected_result)

def test_calc_adjustment_with_zero_parameters(self):
params = np.array([[0, 0, 0, 0, 0, 0, 0]])

result = Weekday.calc_adjustment(params, self.TEST_DATA, ["num"], "date")

# Data should be unchanged when params are 0's
assert np.allclose(result["num"].values, self.TEST_DATA["num"].values)
assert np.allclose(result["den"].values, self.TEST_DATA["den"].values)
assert np.array_equal(result["date"].values, self.TEST_DATA["date"].values)

def test_calc_adjustment(self):
params = np.array([[1, -1, 1, -1, 1, -1, 1]])

result = Weekday.calc_adjustment(params, self.TEST_DATA, ["num"], "date")

print(result["num"].values)
print(result["den"].values)
expected_nums = [
0.36787944,
5.43656366,
1.10363832,
10.87312731,
5,
2.20727665,
19.0279728,
2.94303553,
24.46453646,
3.67879441,
]

# The date and "den" column are unchanged by this function
assert np.allclose(result["num"].values, expected_nums)
assert np.allclose(result["den"].values, self.TEST_DATA["den"].values)
assert np.array_equal(result["date"].values, self.TEST_DATA["date"].values)
7 changes: 5 additions & 2 deletions ansible/templates/covid_act_now-params-prod.json.j2
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"common": {
"export_dir": "/common/covidcast/receiving/covid-act-now",
"export_dir": "./receiving",
"log_filename": "/var/log/indicators/covid_act_now.log"
},
"indicator": {
Expand Down Expand Up @@ -44,5 +44,8 @@
"smoothed_signals": [
]
}
}
},
"delivery": {
"delivery_dir": "/common/covidcast/receiving/covid-act-now"
}
}
1 change: 0 additions & 1 deletion ansible/templates/google_symptoms-params-prod.json.j2
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
"span_length": 14,
"min_expected_lag": {"all": "3"},
"max_expected_lag": {"all": "4"},
"dry_run": true,
"suppressed_errors": [
]
},
Expand Down
5 changes: 4 additions & 1 deletion ansible/templates/quidel_covidtest-params-prod.json.j2
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"common": {
"export_dir": "/common/covidcast/receiving/quidel",
"export_dir": "./receiving",
"log_filename": "/var/log/indicators/quidel_covidtest.log"
},
"indicator": {
Expand Down Expand Up @@ -47,5 +47,8 @@
"covid_ag_smoothed_pct_positive"
]
}
},
"delivery": {
"delivery_dir": "/common/covidcast/receiving/quidel"
}
}
5 changes: 4 additions & 1 deletion ansible/templates/usafacts-params-prod.json.j2
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"common": {
"export_dir": "/common/covidcast/receiving/usa-facts",
"export_dir": "./receiving",
"input_dir": "./input-cache",
"log_filename": "/var/log/indicators/usafacts.log"
},
Expand Down Expand Up @@ -48,5 +48,8 @@
"deaths_7dav_incidence_num",
"deaths_7dav_incidence_prop"]
}
},
"delivery": {
"delivery_dir": "/common/covidcast/receiving/usa-facts"
}
}
1 change: 0 additions & 1 deletion changehc/delphi_changehc/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,5 @@
from . import run
from . import sensor
from . import update_sensor
from . import weekday

__version__ = "0.0.0"
2 changes: 1 addition & 1 deletion changehc/delphi_changehc/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -184,7 +184,7 @@ def run_module(params: Dict[str, Dict[str, Any]]):
file_dict["flu_like"],file_dict["covid_like"],dropdate_dt,"fips")
more_stats = su_inst.update_sensor(
data,
params["common"]["export_dir"]
params["common"]["export_dir"],
)
stats.extend(more_stats)

Expand Down
Loading