U /eT@stddlZddlZddlmZddlZddlmZddlm Z ddl m Z ddl mZddlmZddlmZdd lmZmZmZmZmZmZmZmZmZdd lmZdd lm Z dd l!m"Z"m#Z#m$Z$m%Z%dd l&m'Z'ddl(m)Z)ddl*m+Z+m,Z,m-Z-m.Z.m/Z/e0de1fiZ2ddZ3ddZ4e"eddddddZ5ddZ6ddZ7ddZ8dd Z9Gd!d"d"Z Gd#d$d$e Z:dS)%N)Integral)is_datetime64_any_dtype)Rolling) normalize_arg)tokenize)BlockwiseDepDict)methods) Scalar_Frame_get_divisions_map_partitions_get_meta_map_partitions_maybe_from_pandasapply_and_enforce new_dd_object no_defaultpartitionwise_graph) from_pandas)_maybe_align_partitions)insert_meta_param_descriptionis_dask_collectionis_dataframe_likeis_series_like)unpack_collections)HighLevelGraph)Mapply derived_fromfuncname has_keywordCombinedOutputcCsd}|dk r,t|tr,|jd|kr,t||dk rTt|trT|jd|krTt|dd|||fD}t|}t||dk rt|nd|dk rt|ndfS)NzqPartition size is less than overlapping window size. Try using ``df.repartition`` to increase the partition size.rcSsg|]}|dk r|qSN).0pr!r!:/tmp/pip-unpacked-wheel-dbjnr7gq/dask/dataframe/rolling.py 6sz#_combined_parts..) isinstancershapeNotImplementedErrorrconcatrlen)Z prev_partZ current_partZ next_partbeforeaftermsgpartscombinedr!r!r$_combined_parts's  r0c Osdd|D}|d\}}}dd|D}|||} |dkr@d}t|tjrP|}d} |jddkrv| jd|jd} |r| r|| 9}|dkr| j|dSt|tjr|}|r| r|| 9}| j|| S)NcSsg|]}t|tr|qSr!r&rr"dfr!r!r$r%Cs z!overlap_chunk..rcSs"g|]}t|tr|dn|qSrr1r"argr!r!r$r%Fs)r&datetime timedeltar'iloc) funcr+r,argskwargsdfsr/Zprev_part_lengthZnext_part_lengthoutZ expansionr!r!r$ overlap_chunkBs(   r?T)metaenforce_metadatatransform_divisionsalign_dataframesc st|st|r"t|s"t|dn|}|f|}ttrDtttrXtttj spttj rt |j j j stdn,ttrdkrttrdkstd| dd} | dd} t|st| dk rtf|| } n"dt|} t|f|| } | d | } |rvt|}z t|}Wn4tk rt} zt| d | W5d} ~ XYnXd d |D}t|||| | td d|Dr| dft|tdd |Df| fi}tj| ||d}t|| Sg}g}t|||||| fdd}|D]l}t|t rH||}|!||!|qt"|}t#|\}}|rx|!||$|n |!|qi}d}| %D]:\}}t"|}t#|\}}|$||||<|rd}qt&|drddt'ddD}|(dt)||fdd}|rDt*t+| |f||t,d|}n.|rN| n|}t*t,| |f||d|i}tj| ||d}t-|| S)a] Apply a function to each partition, sharing rows with adjacent partitions. Parameters ---------- func : function The function applied to each partition. If this function accepts the special ``partition_info`` keyword argument, it will recieve information on the partition's relative location within the dataframe. df: dd.DataFrame, dd.Series args, kwargs : Positional and keyword arguments to pass to the function. Positional arguments are computed on a per-partition basis, while keyword arguments are shared across all partitions. The partition itself will be the first positional argument, with all other arguments passed *after*. Arguments can be ``Scalar``, ``Delayed``, or regular Python objects. DataFrame-like args (both dask and pandas) will be repartitioned to align (if necessary) before applying the function; see ``align_dataframes`` to control this behavior. enforce_metadata : bool, default True Whether to enforce at runtime that the structure of the DataFrame produced by ``func`` actually matches the structure of ``meta``. This will rename and reorder columns for each partition, and will raise an error if this doesn't work, but it won't raise if dtypes don't match. before : int, timedelta or string timedelta The rows to prepend to partition ``i`` from the end of partition ``i - 1``. after : int, timedelta or string timedelta The rows to append to partition ``i`` from the beginning of partition ``i + 1``. transform_divisions : bool, default True Whether to apply the function onto the divisions and apply those transformed divisions to the output. align_dataframes : bool, default True Whether to repartition DataFrame- or Series-like args (both dask and pandas) so their divisions align before applying the function. This requires all inputs to have known divisions. Single-partition inputs will be split into multiple partitions. If False, all inputs must have either the same number of partitions or a single partition. Single-partition inputs will be broadcast to every partition of multi-partition inputs. $META See Also -------- dd.DataFrame.map_overlap zMMust have a `DatetimeIndex` when using string offset for `before` and `after`rz*before and after must be positive integerstokenN parent_metazoverlap--zx. If you don't want the partitions to be aligned, and are calling `map_overlap` directly, pass `align_dataframes=False`.cSsg|]}t|tr|qSr!)r&r r2r!r!r$r%s zmap_overlap..css|]}t|tVqdSr )r&r r5r!r!r$ szmap_overlap..cSsg|]}|jdfqSr4)_namer5r!r!r$r%s dependenciesc si}t|\}}||t|\}}||dt|}tt|||D]*\}\}} } ||f} t|| | f|| <qTtj |||gd} t | |S)Nzoverlap-concat-rJ) _get_previous_partitionsupdate_get_nexts_partitionsr enumeratezipZ __dask_keys__r0rfrom_collectionsr) r6dskZprevs_parts_dskprevsZnexts_parts_dsknextsname_aiprevcurrentnextkeygraph)r,r+ divisionsr@r!r$_handle_frame_arguments   z+map_overlap.._handle_frame_argumentTFpartition_infocSsi|]\}}|f||dqS))numberdivisionr!)r"rVr`r!r!r$ szmap_overlap..cs||d|iS)Nr^r!)r^r;r<) orig_funcr!r$r:szmap_overlap..func)rKZ_func_metarK).rrrrr&strpdZ to_timedeltar7r8rindex_meta_nonemptyZ inferred_type TypeErrorr ValueErrorpopcallableAssertionErrorrrr rr allrtuplerrQr r r appendrrextenditemsrrOinsertrrrr?r)r:r3r+r,r@rArBrCr;r<namerFrEer=Zlayerr[Zargs2rKr]r6Zarg2 collectionsZkwargs3simplekvr^rRZkwargs4r!)r,r+r\r@rcr$ map_overlap]sB                         rzc Csi}|j}d}dt||}|rrt|trrg}td|jD]*}||f}tj||f|f||<||q:|dnt|t j rt |j jdd} || krt|g}td|jD]2}||f}t||df||f|f||<||q|dn dg|j}||fS)zE Helper to get the nexts partitions required for the overlap ziPartition size is less than specified window. Try using ``df.repartition`` to increase the partition sizezoverlap-append-rDNrbr)rIrr&rrange npartitionsrheadrpr7r8rfSeriesr\diffr9anyrj_head_timedelta) r3r,rRdf_nameZtimedelta_partition_messageZname_brTrVrZdeltasr!r!r$rN:s0       rNcsi}|jdt||}|rjt|trjdg}t|jdD]*}||f}tj|f|f||<||q:nLt|t j rt |j }|jdd}||kr\|d} dg}t|jdD]}||d} t| || } |||} } | | kr| dkr| || } | d} q||f}tfddt| |dD|df|f||<||qnLdg}t|jdD]6}||f}t|fg|df|f||<||qpn dg|j}||fS)zH Helper to get the previous partitions required for the overlap zoverlap-prepend-NrDrbrcsg|] }|fqSr!r!)r"rxrr!r$r%sz,_get_previous_partitions..)rIrr&rr{r|rtailrpr7r8rfr~r\rr9rmax_tail_timedelta)r3r+rRrUrSrVrZZdivsrZpt_zZpt_iZlbfirstjr!rr$rL_sP       rLcCs||j|j|kS)zReturn rows of ``next_`` whose index is before the last observation in ``current`` + ``after``. Parameters ---------- current : DataFrame next_ : DataFrame after : timedelta Returns ------- overlapped : DataFrame )rgr)rXZnext_r,r!r!r$rsrcstfdd|D}|S)a4Return the concatenated rows of each dataframe in ``prevs`` whose index is after the first observation in ``current`` - ``before``. Parameters ---------- current : DataFrame prevs : list of DataFrame objects before : timedelta Returns ------- overlapped : DataFrame cs$g|]}||jjkqSr!)rgmin)r"rWr+rXr!r$r%sz#_tail_timedelta..)rr))rSrXr+selectedr!rr$rsrc@s6eZdZdZd0ddZddZed d Zed d Z d dZ e e ddZ e e ddZe e ddZe e ddZe e ddZe e ddZe e ddZe e d1ddZe e d2d d!Ze e d"d#Ze e d$d%Ze e d&d'Ze e d3d)d*Ze e d4d,d-ZeZd.d/ZdS)5rz%Provides rolling window calculations.NFrcCsP||_||_||_||_||_||_|jjf|t |jt rFdnd|_ dS)Nfreq) objwindow min_periodscenteraxiswin_typerdrolling_rolling_kwargsr&int _win_type)selfrrrrrrr!r!r$__init__szRolling.__init__cCs|j|j|j|j|jdSNrrrrrrrr!r!r$rs zRolling._rolling_kwargscCs,|jdkp*t|jtr |jdkp*|jjdkS)zm Indicator for whether the object has a single partition (True) or multiple (False). )rDcolumnsrD)rr&rrrr|rr!r!r$_has_single_partitions   zRolling._has_single_partitioncOs|jf|}t||||Sr )rgetattr)r3rolling_kwargsrtr;r<rr!r!r$pandas_rolling_methods zRolling.pandas_rolling_methodcOs|}|j|jj||f||}|jrL|jj|j||f|||d|S|jrl|jd}|j|d}n*|jdkrt |j}d}n|jd}d}t |j|j||||f|||d|S)N)rEr@rDrr) rrrrhrZmap_partitionsrrrrfZ Timedeltarz)r method_namer;r<rr@r+r,r!r!r$ _call_methods\      zRolling._call_methodcCs |dS)Ncountrrr!r!r$rsz Rolling.countcCs |dS)Ncovrrr!r!r$r!sz Rolling.covcCs |dS)Nsumrrr!r!r$r%sz Rolling.sumcCs |dS)Nmeanrrr!r!r$r)sz Rolling.meancCs |dS)Nmedianrrr!r!r$r-szRolling.mediancCs |dS)Nrrrr!r!r$r1sz Rolling.mincCs |dS)Nrrrr!r!r$r5sz Rolling.maxrDcCs|jdddS)NstdrDddofrrrr!r!r$r9sz Rolling.stdcCs|jdddS)NvarrDrrrr!r!r$r=sz Rolling.varcCs |dS)Nskewrrr!r!r$rAsz Rolling.skewcCs |dS)Nkurtrrr!r!r$rEsz Rolling.kurtcCs |d|S)Nquantiler)rrr!r!r$rIszRolling.quantilecythonc Cspi}|p i}|pd}|jjd}t|jdr:t||d}|dkrTt|jjd}|j d|f|||d|S)Nr!rengine)r engine_kwargsrawr)rr;r<) rrdrrrdictinspect signature parametersr) rr:rrrr;r<Z compat_kwargsr@r!r!r$rMs$   z Rolling.applyr!cKs&|dkr i}|jd|f||d|S)Nagg)r;r<r)rr:r;r<kwdsr!r!r$ aggregatefszRolling.aggregatec CsJdd}|}|j|d<|j|d<ddddt||d DS) NcSs |\}}dddddd}||S)NrrDrrr!)itemrxry_orderr!r!r$orderoszRolling.__repr__..orderrrz Rolling [{}],css(|] \}}|dk r|d|VqdS)N=r!)r"rxryr!r!r$rH~sz#Rolling.__repr__..)rZ)rrrformatjoinsortedrr)rrrr!r!r$__repr__ns    zRolling.__repr__)NNFNr)rD)rD)NrNNN)r!N)__name__ __module__ __qualname____doc__rrpropertyr staticmethodrrr pd_Rollingrrrrrrrrrrrrrrrrr!r!r!r$rs`    (              rcs>eZdZd fdd ZedddddZfd d ZZS) RollingGroupbyNFrc s|j|_|j|_|j}|jdk rnt|jtr6|jg}n t|j}t|jtrZ||jn | |j||}t j ||||||ddSr) _groupby_kwargsZ_slice_groupby_slicerr&relistZbyrprqsuperr) rgroupbyrrrrrrZ sliced_plus __class__r!r$rs&       zRollingGroupby.__init__groupby_kwargs groupby_slicec Os<|jf|}|r||}|jf|}t||||jddS)Nrb)level)rrrZ sort_index) r3rrtrrr;r<rrr!r!r$rs  z$RollingGroupby.pandas_rolling_methodcs"tj|f||j|jd|S)Nr)rrrr)rrr;r<rr!r$rszRollingGroupby._call_method)NNFNr)rrrrrrr __classcell__r!r!rr$rs!r);r7rZnumbersrZpandasrfZpandas.api.typesrZpandas.core.windowrrZdask.array.corerZ dask.baserZdask.blockwiserZdask.dataframerZdask.dataframe.corer r r r r rrrrZdask.dataframe.iorZdask.dataframe.multirZdask.dataframe.utilsrrrrZ dask.delayedrZdask.highlevelgraphrZ dask.utilsrrrrrtyperorr0r?rzrNrLrrrr!r!r!r$sB       ,    ]%BA