`send`ing into `zip`ped generators with `yield from`

223 Views Asked by At

I am using Python 3.6 where you can nicely zip individual generators of one kind to get a multidimensional generator of the same kind. Take the following example, where get_random_sequence is a generator that yields an infinite random number sequence to simulate one individual asset at a stock market.

import random
from typing import Iterator

def get_random_sequence(start_value: float) -> Iterator[float]:
    value = start_value
    yield value

    while True:
        r = random.uniform(-.1, .1) * value
        value = value + r
        yield value

g = get_random_sequence(1.)
a = [next(g) for _ in range(5)]
print(a)
# [1.0, 1.015821868415922, 1.051470250712725, 0.9827564500218019, 0.9001851912863]

This generator can be easily extended with Python's zip function and yield from to generate successive asset values at a simulated market with an arbitrary number of assets.

def get_market(no_assets: int = 10) -> Iterator[Sequence[float]]:
    rg = tuple(get_random_sequence(random.uniform(10., 60.)) for _ in range(no_assets))
    yield from zip(*rg)

gm = get_market(2)
b = [next(gm) for _ in range(5)]
print(b)
# [(55.20719435959121, 54.15552382961163), (51.64409510285255, 53.6327489348457), (50.16517363363749, 52.92881727359184), (48.8692976247231, 52.7090801870517), (52.49414777987645, 49.733746261206036)]

What I like about this approach is the use of yield from to avoid a while True: loop in which a tuple of n assets would have to be constructed explicitly.

My question is: Is there a way to apply yield from in a similar manner when the zipped generators receive values over send()?

Consider the following generator that yields the ratio of successive values in an infinite sequence.

from typing import Optional, Generator

def ratio_generator() -> Generator[float, Optional[float], None]:
    value_last = yield
    value = yield
    while True:
        ratio = 0. if value_last == 0. else value / value_last
        value_last = value
        value = yield ratio

gr = ratio_generator()
next(gr)    # move to the first yield
g = get_random_sequence(1.)
a = []
for _v in g:
    _r = gr.send(_v)
    if _r is None:
        # two values are required for a ratio
        continue
    a.append(_r)
    if len(a) >= 5:
        break
print(a)
# [1.009041186223442, 0.9318419861800313, 1.0607677437816718, 0.9237896996817375, 0.9759635921282439]

The best way to "zip" this generator I could come up with, unfortunately, does not involve yield from at all... but instead the ugly while True: solution mentioned above.

def ratio_generator_multiple(no_values: int) -> Generator[Sequence[float], Optional[Sequence[float]], None]:
    gs = tuple(ratio_generator() for _ in range(no_values))
    for each_g in gs:
        next(each_g)

    values = yield
    ratios = tuple(g.send(v) for g, v in zip(gs, values))

    while True: # :(
        values = yield None if None in ratios else ratios
        ratios = tuple(g.send(v) for g, v in zip(gs, values))

rgm = ratio_generator_multiple(2)
next(rgm)    # move to the first yield
gm = get_market(2)
b = []
for _v in gm:
    _r = rgm.send(_v)
    if _r is None:
        # two values are required for a ratio
        continue
    b.append(_r)
    if len(b) >= 5:
        break
print(b)
# [(1.0684036496767984, 1.0531433541856687), (1.0279604693226763, 1.0649271401851732), (1.0469406709985847, 0.9350856571355237), (0.9818403001921499, 1.0344633443394962), (1.0380945284830183, 0.9081599684720663)]

Is there a way to do something like values = yield from zip(*(g.send(v) for g, v in zip(generators, values))) so I can still use yield from on the zipped generators without a while True:? (The given example doesn't work, because it doesn't refresh the values on the right-hand side with the values on the left-hand side.)

I realize that this is more of an aesthetic problem. It would still be nice to know though...

0

There are 0 best solutions below