Rustは本当に動作が高速なのか？　Pythonとの比較で分かる、Rustのパフォーマンス特性：WebエンジニアからみたRust（2）

def calc_batch_ndarray(calc_strategy, average_length) -> np.ndarray:
    before_read = datetime.utcnow()  # データ読み込み前の時刻記録
    nums = pd.read_csv("../data/time_series.csv")['value'].values  # データ一括読み込み
    after_read = datetime.utcnow()  # データ読み込み後の時刻記録
    moving_averages = calc_strategy(nums, average_length)  # 関数を用いて移動平均計算
    after_calc = datetime.utcnow()  # 移動平均計算後の時刻記録
    print(f"移動平均計算に使用した関数：{calc_strategy}")
    print(f"移動平均の長さ：{average_length}")
    print(f"移動平均の最後の要素：{moving_averages[-1]}")
    print(f"csvロードにかかった時間：{after_read - before_read }秒")
    print(f"移動平均計算にかかった時間：{after_calc - after_read}秒")
    print(f"配列のメモリ使用量(参考)：{moving_averages.nbytes/1e6}MB")
    print(f"プロセスメモリ使用量(参考)：{process_memory_usage_mb()}MB")
    return moving_averages
def moving_average_batch_numpy(nums: np.ndarray, average_length: int) -> np.ndarray:  # numpy arrayを使う、convolve APIを使う
    assert len(nums) - average_length + 1 > 0
    return np.convolve(nums, np.ones(average_length), 'valid') / average_length
if __name__ == "__main__":
    ma1 = calc_batch_ndarray(calc_strategy=moving_average_batch_numpy, average_length=7)

fn moving_average_batch_online(nums: &[f64], average_length: usize) -> anyhow::Result<Vec<f64>> {
    let size = nums.len() as i64 - average_length as i64 + 1; // 出力される移動平均の数列のサイズ
    if size <= 0 {
        // サイズが0以下ならばエラー値を関数の戻り値として返す
        return Err(anyhow::anyhow!(
            "average length must be less than nums array length"
        ));
    }
    let mut res = Vec::with_capacity(nums.len());
    // 直近N個のデータの総和を計算する、最初のデータは素直に総和をとる
    res.push(nums[0..average_length].iter().sum::<f64>());
    // 最初のデータ以外は前の総和から新しいデータを足して、古いデータを引くことで計算し、計算量を減らす
    for i in average_length..nums.len() {
        res.push(nums[i] as f64 - nums[i - average_length] as f64 + res[i - average_length])
    }
    // 出力するデータをNで割る、後で割るのは計算誤差を小さくするため
    for i in 0..(nums.len() - average_length + 1) {
        res[i] /= average_length as f64;
    }
    Ok(res)
}
fn main() -> anyhow::Result<()> {
    let _ma = calc_batch(moving_average_batch_online, 5000)?;
    Ok(())
}

@numba.jit(nopython=True)
# numpy arrayを使う、オンラインアルゴリズムを使用、Numbaを使う
def moving_average_batch_numpy_numba_online(nums: np.ndarray, average_length: int) -> np.ndarray:
    assert len(nums) - average_length + 1 > 0
    N_i = nums.shape[0]
    res = np.empty_like(nums, dtype=np.float64)
    res[average_length - 1] = np.sum(nums[:average_length])
    for i in range(average_length, N_i):
        res[i] = nums[i] - nums[i - average_length] + res[i - 1]
    for i in range(average_length-1, N_i):
        res[i] = res[i] / average_length
    return res[average_length-1:]
  
if __name__ == "__main__":
    ma8 = calc_batch_ndarray(calc_strategy=moving_average_batch_numpy_numba_online, average_length=5000)

#[derive(Debug, Clone)]
pub struct MovingAverage {
    period: usize,
    sum: f64,
    deque: std::collections::VecDeque<f64>,
}
impl MovingAverage {
    pub fn new(period: usize) -> Self {
        Self {
            period,
            sum: 0.0,
            deque: std::collections::VecDeque::new(),
        }
    }
    pub fn latest(&mut self, new_val: f64) -> Option<f64> {
        self.deque.push_back(new_val);
        let old_val = match self.deque.len() > self.period {
            true => self.deque.pop_front().unwrap(),
            false => 0.0,
        };
        self.sum += new_val - old_val;
        match self.deque.len() == self.period {
            true => Some(self.sum / self.period as f64),
            false => None,
        }
    }
}
fn calc_stream(average_length: usize) -> anyhow::Result<Vec<f64>> {
    let before_read = chrono::Utc::now();
    let csv_path = get_csv_path("data/time_series.csv");
    let mut csv_reader = csv::Reader::from_path(csv_path)?;
    let mut ma = MovingAverage::new(average_length);
    let moving_averages = csv_reader
        .deserialize::<f64>()
        .filter_map(|row_result| row_result.ok())
        .filter_map(|new_val| ma.latest(new_val))
        .collect::<Vec<_>>();
    let after_calc = chrono::Utc::now();
    println!("移動平均の長さ：{}", average_length);
    println!(
        "移動平均の最後の要素：{:?}",
        moving_averages[moving_averages.len() - 1]
    );
    let total = after_calc - before_read;
    println!(
        "計算にかかった時間：{:?}秒",
        total.num_nanoseconds().unwrap() as f64 / 1e9
    );
    println!(
        "Vecの使用メモリ量(参考)：{:?}MB",
        std::mem::size_of_val(&*moving_averages) as f64 / 1e6
    );
    println!(
        "プロセスの使用メモリ量(参考)：{:?}MB",
        psutil::process::Process::new(std::process::id())
            .unwrap()
            .memory_info()
            .unwrap()
            .rss() as f64
            / 1e6
    );
    Ok(moving_averages)
}
fn main() -> anyhow::Result<()> {
    let ma4 = calc_stream(5000)?;
    Ok(())
}

class MovingAveragePython(object):
    def __init__(self, period):
        self.sum = 0
        self.period = period
        self.deque = collections.deque()
    def latest(self, new_val):
        self.deque.append(new_val)
        old_val = self.deque.popleft() if len(self.deque) > self.period else 0.0
        self.sum += new_val - old_val
        return self.sum / self.period if len(self.deque) == self.period else None
def calc_stream(constructor, average_length) -> np.ndarray:
    before_read = datetime.utcnow()  # データ読み込み前の時刻記録
    moving_averages = []
    ma = constructor(average_length)
    with open("../data/time_series.csv") as f:
        reader = csv.reader(f)
        next(reader) # ヘッダをskipする
        for row in reader:
            num = float(row[0])
            moving_averages.append(ma.latest(num))
    after_calc = datetime.utcnow()  # 移動平均計算後の時刻記録
    print(f"移動平均の長さ：{average_length}")
    print(f"移動平均の最後の要素：{moving_averages[-1]}")
    print(f"計算にかかった時間：{after_calc - before_read}秒")
    print(f"リストのメモリ使用量(参考)：{sys.getsizeof(moving_averages)/1e6}MB")
    print(f"プロセスメモリ使用量(参考)：{process_memory_usage_mb()}MB")
    return moving_averages
if __name__ == "__main__":
    ma3 = calc_stream(constructor=MovingAveragePython, average_length=5000)