python
pip install nupic
python
from datetime import datetime
from nupic.frameworks.opf.model_factory import ModelFactory
from nupic.data.inference_shifter import InferenceShifter
from nupic.encoders.date import DateEncoder
from nupic.encoders.random_distributed_scalar import \
RandomDistributedScalarEncoder
from nupic.output_store import OutputStore
from nupic.encoders.multi import MultiEncoder
data_directory = "<path_to_data_directory>"
prediction_length = 5
input_data_path = data_directory + "input_data.csv"
output_data_path = data_directory + "output_data.csv"
model_params_path = data_directory + "model_params.py"
date_encoder = DateEncoder(
timeOfDay=date_string,
name="date_encoder",
startOfDay=0,
endOfDay=23.99,
clipInput=True,
forced=True
)
num_categories = 100
random_distributed_encoder = RandomDistributedScalarEncoder(
numBuckets=num_categories,
bitsEachBucket=[2]
)
encoder = MultiEncoder()
encoder.addEncoder(date_encoder)
encoder.addEncoder(random_distributed_encoder)
python
def read_csv_data(path):
data = []
with open(path, "r") as file:
for line in file:
data.append(float(line.strip()))
return data
def process_data(data):
processed_data = []
for i in range(len(data)):
processed_data.append({
"date": datetime(year=2022, month=1, day=i + 1),
"value": data[i]
})
return processed_data
def convert_to_nupic_format(data):
converted_data = []
for item in data:
converted_data.append(encoder.encode(item["date"]) + encoder.encode(item["value"]))
return converted_data
raw_data = read_csv_data(input_data_path)
processed_data = process_data(raw_data)
converted_data = convert_to_nupic_format(processed_data)
python
model_params = ModelFactory.create(model_params_path)
model = ModelFactory.create(model_params)
model.enableInference({"predictedField": "value"})
for item in converted_data:
modelInput = {"input": item}
modelInputWithTime = {"input": item, "timestamp": datetime.now()}
result = model.run(modelInputWithTime)
output_data.append(result.inferences["multiStepBestPredictions"][prediction_length-1]["value"])
def save_predictions(output_data):
with open(output_data_path, "w") as file:
for value in output_data:
file.write(str(value) + "
")
save_predictions(output_data)
