Predicting Sunspot Activity with CnosDB and a TensorFlow 1DConv‑LSTM Model

Sunspot numbers, a key indicator of solar activity, exhibit a roughly 11‑year cycle and have shown a recent declining trend, making accurate forecasting important for space weather research.

The monthly mean sunspot number (MSSN) dataset from the SILSO website (1749‑2023) is downloaded as SN_m_tot_V2.0.csv and loaded with pandas:

import pandas as pd df = pd.read_csv("SN_m_tot_V2.0.csv", sep=";", header=None) df.columns = ["year", "month", "date_fraction", "mssn", "standard_deviation", "observations", "marker"] df["year"] = df["year"].astype(str) df["month"] = df["month"].astype(str) df["date"] = df["year"] + "-" + df["month"] print(df.head())

The data are stored in CnosDB, an open‑source distributed time‑series database. After launching CnosDB with Docker, a table sunspot is created via the CLI:

public ❯ CREATE TABLE sunspot ( date STRING, mssn DOUBLE, );

Python interaction uses the CnosDB connector:

# install connector pip install -U cnos-connector from cnosdb_connector import connect conn = connect(url="http://127.0.0.1:31001/", user="root", password="") cursor = conn.cursor() # create database and table conn.create_database("tf_demo") conn.switch_database("tf_demo") cursor.execute("CREATE TABLE sunspot (date STRING, mssn DOUBLE);") # write dataframe to CnosDB conn.write_dataframe(df, "sunspot", ["date", "mssn"])

Data are read back for modeling:

df = pd.read_sql("select * from sunspot;", conn) print(df.head())

The dataset is split into training and testing sets (80/20) and transformed into a sliding‑window format for time‑series learning:

import numpy as np time_index = np.array(df['date']) data = np.array(df['mssn']) SPLIT_RATIO = 0.8 split_index = int(SPLIT_RATIO * data.shape[0]) train_data = data[:split_index] train_time = time_index[:split_index] test_data = data[split_index:] test_time = time_index[split_index:] WINDOW_SIZE = 60 BATCH_SIZE = 32 SHUFFLE_BUFFER = 1000 import tensorflow as tf def ts_data_generator(data, window_size, batch_size, shuffle_buffer): ds = tf.data.Dataset.from_tensor_slices(data) ds = ds.window(window_size + 1, shift=1, drop_remainder=True) ds = ds.flat_map(lambda w: w.batch(window_size + 1)) ds = ds.shuffle(shuffle_buffer).map(lambda w: (w[:-1], w[-1])) ds = ds.batch(batch_size).prefetch(1) return ds tensor_train_data = tf.expand_dims(train_data, axis=-1) tensor_train_dataset = ts_data_generator(tensor_train_data, WINDOW_SIZE, BATCH_SIZE, SHUFFLE_BUFFER) tensor_test_data = tf.expand_dims(test_data, axis=-1) tensor_test_dataset = ts_data_generator(tensor_test_data, WINDOW_SIZE, BATCH_SIZE, SHUFFLE_BUFFER)

A 1D convolution followed by two LSTM layers and dense output layers constitutes the forecasting model:

model = tf.keras.models.Sequential([ tf.keras.layers.Conv1D(filters=128, kernel_size=3, strides=1, input_shape=[None, 1]), tf.keras.layers.MaxPool1D(pool_size=2, strides=1), tf.keras.layers.LSTM(128, return_sequences=True), tf.keras.layers.LSTM(64, return_sequences=True), tf.keras.layers.Dense(132, activation="relu"), tf.keras.layers.Dense(1) ]) optimizer = tf.keras.optimizers.Adam(learning_rate=1e-3) model.compile(loss="mse", optimizer=optimizer, metrics=["mae"]) history = model.fit(tensor_train_dataset, epochs=20, validation_data=tensor_test_dataset)

Training loss and validation loss are plotted to assess convergence:

import matplotlib.pyplot as plt plt.plot(history.history['loss']) plt.plot(history.history['val_loss']) plt.title('model loss') plt.ylabel('loss') plt.xlabel('epoch') plt.legend(['train', 'test'], loc='upper left') plt.show()

After training, the model forecasts MSSN values; mean absolute error on the test set is reported (≈24.68):

def model_forecast(model, data, window_size): ds = tf.data.Dataset.from_tensor_slices(data) ds = ds.window(window_size, shift=1, drop_remainder=True) ds = ds.flat_map(lambda w: w.batch(window_size)) ds = ds.batch(32).prefetch(1) forecast = model.predict(ds) return forecast rnn_forecast = model_forecast(model, data[..., np.newaxis], WINDOW_SIZE) rnn_forecast = rnn_forecast[split_index - WINDOW_SIZE:-1, -1, 0] error = tf.keras.metrics.mean_absolute_error(test_data, rnn_forecast).numpy() print(error) # 24.676455

Finally, the predicted series is plotted against the ground‑truth series to visualize performance:

plt.plot(test_data) plt.plot(rnn_forecast) plt.title('MSSN Forecast') plt.ylabel('MSSN') plt.xlabel('Month') plt.legend(['Ground Truth', 'Predictions'], loc='upper right') plt.show()