Эффективное вычисление среднего значения списка в Haskell

Question

Я разработал функцию для вычисления среднего значения списка.Хотя это работает нормально, но я думаю, что это может быть не лучшим решением, потому что для этого требуется две функции, а не одна.Можно ли выполнить эту работу только с одной рекурсивной функцией?

calcMeanList (x:xs) = doCalcMeanList (x:xs) 0 0

doCalcMeanList (x:xs) sum length =  doCalcMeanList xs (sum+x) (length+1)
doCalcMeanList [] sum length = sum/length

Answer 1

Ваше решение хорошо, использование двух функций не хуже одной.Тем не менее, вы можете поместить хвостовую рекурсивную функцию в предложение where.

Но если вы хотите сделать это в одной строке:

calcMeanList = uncurry (/) . foldr (\e (s,c) -> (e+s,c+1)) (0,0)

Answer 2

Самое лучшее, что вы можете сделать, это эта версия :

import qualified Data.Vector.Unboxed as U

data Pair = Pair {-# UNPACK #-}!Int {-# UNPACK #-}!Double

mean :: U.Vector Double -> Double
mean xs = s / fromIntegral n
  where
    Pair n s       = U.foldl' k (Pair 0 0) xs
    k (Pair n s) x = Pair (n+1) (s+x)

main = print (mean $ U.enumFromN 1 (10^7))

Она сливается с оптимальным циклом в Core (лучший Haskell, который вы можете написать):

main_$s$wfoldlM'_loop :: Int#
                              -> Double#
                              -> Double#
                              -> Int#
                              -> (# Int#, Double# #)    
main_$s$wfoldlM'_loop =
  \ (sc_s1nH :: Int#)
    (sc1_s1nI :: Double#)
    (sc2_s1nJ :: Double#)
    (sc3_s1nK :: Int#) ->
    case ># sc_s1nH 0 of _ {
      False -> (# sc3_s1nK, sc2_s1nJ #);
      True ->
        main_$s$wfoldlM'_loop
          (-# sc_s1nH 1)
          (+## sc1_s1nI 1.0)
          (+## sc2_s1nJ sc1_s1nI)
          (+# sc3_s1nK 1)
    }

И следующая сборка:

Main_mainzuzdszdwfoldlMzqzuloop_info:
.Lc1pN:
        testq %r14,%r14
        jg .Lc1pQ
        movq %rsi,%rbx
        movsd %xmm6,%xmm5
        jmp *(%rbp)
.Lc1pQ:
        leaq 1(%rsi),%rax
        movsd %xmm6,%xmm0
        addsd %xmm5,%xmm0
        movsd %xmm5,%xmm7
        addsd .Ln1pS(%rip),%xmm7
        decq %r14
        movsd %xmm7,%xmm5
        movsd %xmm0,%xmm6
        movq %rax,%rsi
        jmp Main_mainzuzdszdwfoldlMzqzuloop_info

На основе Data.Vector.Например,

$ ghc -Odph --make A.hs -fforce-recomp
[1 of 1] Compiling Main             ( A.hs, A.o )
Linking A ...
$ time ./A
5000000.5
./A  0.04s user 0.00s system 93% cpu 0.046 total

См. Эффективные реализации в пакете статистики .

Answer 3

Когда я увидел ваш вопрос, я сразу подумал: "Вы хотите сбросить там!"StackOverflow и этот ответ имеет очень эффективное решение, которое вы можете протестировать в интерактивной среде, такой как GHCi:

import Data.List

let avg l = let (t,n) = foldl' (\(b,c) a -> (a+b,c+1)) (0,0) l 
            in realToFrac(t)/realToFrac(n)

avg ([1,2,3,4]::[Int])
2.5
avg ([1,2,3,4]::[Double])
2.5

Answer 4

Для тех, кому любопытно узнать, как будет выглядеть подход кодера и Ассафа в Haskell, вот один перевод:

avg [] = 0
avg x@(t:ts) = let xlen = toRational $ length x
                   tslen = toRational $ length ts
                   prevAvg = avg ts
               in (toRational t) / xlen + prevAvg * tslen / xlen

Этот способ гарантирует, что на каждом шаге правильно рассчитано «среднее значение», но это происходит за счет целого набора избыточных умножений / делений на длины и очень неэффективных вычислений длины на каждом шаге. Ни один опытный Хаскеллер не напишет это так.

Только немного лучше:

avg2 [] = 0
avg2 x = fst $ avg_ x
    where 
      avg_ [] = (toRational 0, toRational 0)
      avg_ (t:ts) = let
           (prevAvg, prevLen) = avg_ ts
           curLen = prevLen + 1
           curAvg = (toRational t) / curLen + prevAvg * prevLen / curLen
        in (curAvg, curLen)

Это позволяет избежать повторного расчета длины. Но для этого требуется вспомогательная функция, которой и старается избегать оригинальный постер. И это все еще требует целой связки отмены сроков.

Чтобы избежать сокращения длин, мы можем просто собрать сумму и длину и разделить в конце:

avg3 [] = 0
avg3 x = (toRational total) / (toRational len)
    where 
      (total, len) = avg_ x
      avg_ [] = (0, 0)
      avg_ (t:ts) = let 
          (prevSum, prevLen) = avg_ ts
       in (prevSum + t, prevLen + 1)

И это может быть гораздо более кратко записано как свёртка:

avg4 [] = 0
avg4 x = (toRational total) / (toRational len)
    where
      (total, len) = foldr avg_ (0,0) x
      avg_ t (prevSum, prevLen) = (prevSum + t, prevLen + 1)

, что может быть дополнительно упрощено согласно сообщениям выше.

Фолд - действительно путь сюда.

Answer 5

Хотя я не уверен, было бы «лучше» написать это в одной функции, это можно сделать следующим образом:

Если вы заранее знаете длину (давайте назовем ее «n»), ее легко - вы можете вычислить, сколько каждое значение «добавляет» к среднему; это будет значение / длина. Поскольку avg (x1, x2, x3) = сумма (x1, x2, x3) / длина = (x1 + x2 + x3) / 3 = x1 / 3 + x2 / 3 + x2 / 3

Если вы не знаете длину заранее, это немного сложнее:

Допустим, мы используем список {x1, x2, x3}, не зная его n = 3.

первая итерация будет просто x1 (так как мы предполагаем, что ее n = 1) вторая итерация добавит x2 / 2 и разделит существующее среднее на 2, так что теперь мы имеем x1 / 2 + x2 / 2

после третьей итерации у нас n = 3, и мы хотели бы иметь x1 / 3 + x2 / 3 + x3 / 3, но мы имеем x1 / 2 + x2 / 2

поэтому нам нужно было бы умножить на (n-1) и разделить на n, чтобы получить x1 / 3 + x2 / 3, и к этому мы просто добавляем текущее значение (x3), деленное на n, чтобы в итоге получить x1 / 3 + х2 / 3 + х3 / 3

В целом:

учитывая среднее (среднее арифметическое - avg) для n-1 элементов, если вы хотите добавить один элемент (newval) к среднему, ваше уравнение будет:

avg * (n-1) / n + newval / n. Уравнение можно доказать математически с помощью индукции.

Надеюсь, это поможет.

* обратите внимание, что это решение менее эффективно, чем простое суммирование переменных и деление на общую длину, как в вашем примере.

Answer 6

Чтобы продолжить ответ Дона 2010 года, на GHC 8.0.2 мы можем сделать намного лучше.Сначала давайте попробуем его версию.

module Main (main) where

import System.CPUTime.Rdtsc (rdtsc)
import Text.Printf (printf)
import qualified Data.Vector.Unboxed as U

data Pair = Pair {-# UNPACK #-}!Int {-# UNPACK #-}!Double

mean' :: U.Vector Double -> Double
mean' xs = s / fromIntegral n
  where
    Pair n s       = U.foldl' k (Pair 0 0) xs
    k (Pair n s) x = Pair (n+1) (s+x)

main :: IO ()
main = do
  s <- rdtsc
  let r = mean' (U.enumFromN 1 30000000)
  e <- seq r rdtsc
  print (e - s, r)

Это дает нам

[nix-shell:/tmp]$ ghc -fforce-recomp -O2 MeanD.hs -o MeanD && ./MeanD +RTS -s
[1 of 1] Compiling Main             ( MeanD.hs, MeanD.o )
Linking MeanD ...
(372877482,1.50000005e7)
     240,104,176 bytes allocated in the heap
           6,832 bytes copied during GC
          44,384 bytes maximum residency (1 sample(s))
          25,248 bytes maximum slop
             230 MB total memory in use (0 MB lost due to fragmentation)

                                     Tot time (elapsed)  Avg pause  Max pause
  Gen  0         1 colls,     0 par    0.000s   0.000s     0.0000s    0.0000s
  Gen  1         1 colls,     0 par    0.006s   0.006s     0.0062s    0.0062s

  INIT    time    0.000s  (  0.000s elapsed)
  MUT     time    0.087s  (  0.087s elapsed)
  GC      time    0.006s  (  0.006s elapsed)
  EXIT    time    0.006s  (  0.006s elapsed)
  Total   time    0.100s  (  0.099s elapsed)

  %GC     time       6.2%  (6.2% elapsed)

  Alloc rate    2,761,447,559 bytes per MUT second

  Productivity  93.8% of total user, 93.8% of total elapsed

Однако код прост: в идеале не должно быть необходимости в векторе: оптимальный код должен быть возможен при простом встраиваниигенерация списка.К счастью, GHC может сделать это для нас [0].

module Main (main) where

import System.CPUTime.Rdtsc (rdtsc)
import Text.Printf (printf)
import Data.List (foldl')

data Pair = Pair {-# UNPACK #-}!Int {-# UNPACK #-}!Double

mean' :: [Double] -> Double
mean' xs = v / fromIntegral l
  where
    Pair l v = foldl' f (Pair 0 0) xs
    f (Pair l' v') x = Pair (l' + 1) (v' + x)

main :: IO ()
main = do
  s <- rdtsc
  let r = mean' $ fromIntegral <$> [1 :: Int .. 30000000]
      -- This is slow!
      -- r = mean' [1 .. 30000000]
  e <- seq r rdtsc
  print (e - s, r)

Это дает нам:

[nix-shell:/tmp]$ ghc -fforce-recomp -O2 MeanD.hs -o MeanD && ./MeanD +RTS -s
[1 of 1] Compiling Main             ( MeanD.hs, MeanD.o )
Linking MeanD ...
(128434754,1.50000005e7)
         104,064 bytes allocated in the heap
           3,480 bytes copied during GC
          44,384 bytes maximum residency (1 sample(s))
          17,056 bytes maximum slop
               1 MB total memory in use (0 MB lost due to fragmentation)

                                     Tot time (elapsed)  Avg pause  Max pause
  Gen  0         0 colls,     0 par    0.000s   0.000s     0.0000s    0.0000s
  Gen  1         1 colls,     0 par    0.000s   0.000s     0.0000s    0.0000s

  INIT    time    0.000s  (  0.000s elapsed)
  MUT     time    0.032s  (  0.032s elapsed)
  GC      time    0.000s  (  0.000s elapsed)
  EXIT    time    0.000s  (  0.000s elapsed)
  Total   time    0.033s  (  0.032s elapsed)

  %GC     time       0.1%  (0.1% elapsed)

  Alloc rate    3,244,739 bytes per MUT second

  Productivity  99.8% of total user, 99.8% of total elapsed

[0]: обратите внимание, как я должен был отобразить fromIntegral: без этого,GHC не может устранить [Double], и решение намного медленнее.Это несколько грустно: я не понимаю, почему GHC не может встроить / решает, что без этого не нужно.Если у вас есть настоящая коллекция дробей, то этот хак вам не подойдет, и вектор все равно может понадобиться.