ENH: hack job but panel resampling with NumPy function works, close #1149

Author: wesm

pandas/core/groupby.py

@@ -1421,7 +1421,7 @@ def _aggregate_multiple_funcs(self, arg):
             try:
                 colg = SeriesGroupBy(obj[col], column=col,
                                      grouper=self.grouper)
-                results.append(colg.agg(arg))
+                results.append(colg.aggregate(arg))
                 keys.append(col)
             except (TypeError, GroupByError):
                 pass
@@ -1437,7 +1437,7 @@ def _aggregate_generic(self, func, *args, **kwargs):
         obj = self._obj_with_exclusions
 
         result = {}
-        if axis == 0:
+        if axis != obj._het_axis:
             try:
                 for name in self.indices:
                     data = self.get_group(name, obj=obj)
@@ -1453,19 +1453,10 @@ def _aggregate_generic(self, func, *args, **kwargs):
                     wrapper = lambda x: func(x, *args, **kwargs)
                     result[name] = data.apply(wrapper, axis=axis)
 
-        result_index = self.grouper.levels[0]
-
-        if result:
-            if axis == 0:
-                result = DataFrame(result, index=obj.columns,
-                                   columns=result_index).T
-            else:
-                result = DataFrame(result, index=obj.index,
-                                   columns=result_index)
-        else:
-            result = DataFrame(result)
+        return self._wrap_generic_output(result, obj)
 
-        return result
+    def _wrap_aggregated_output(self, output, names=None):
+        raise NotImplementedError
 
     def _aggregate_item_by_item(self, func, *args, **kwargs):
         # only for axis==0
@@ -1477,7 +1468,7 @@ def _aggregate_item_by_item(self, func, *args, **kwargs):
             try:
                 colg = SeriesGroupBy(obj[item], column=item,
                                      grouper=self.grouper)
-                result[item] = colg.agg(func, *args, **kwargs)
+                result[item] = colg.aggregate(func, *args, **kwargs)
             except (ValueError, TypeError):
                 cannot_agg.append(item)
                 continue
@@ -1488,39 +1479,21 @@ def _aggregate_item_by_item(self, func, *args, **kwargs):
 
         return DataFrame(result, columns=result_columns)
 
-    def _wrap_aggregated_output(self, output, names=None):
-        agg_axis = 0 if self.axis == 1 else 1
-        agg_labels = self._obj_with_exclusions._get_axis(agg_axis)
-
-        if len(output) == len(agg_labels):
-            output_keys = agg_labels
+    def _decide_output_index(self, output, labels):
+        if len(output) == len(labels):
+            output_keys = labels
         else:
             output_keys = sorted(output)
             try:
                 output_keys.sort()
             except Exception:  # pragma: no cover
                 pass
 
-            if isinstance(agg_labels, MultiIndex):
+            if isinstance(labels, MultiIndex):
                 output_keys = MultiIndex.from_tuples(output_keys,
-                                                     names=agg_labels.names)
-
-        if not self.as_index:
-            result = DataFrame(output, columns=output_keys)
-            group_levels = self.grouper.get_group_levels()
-            zipped = zip(self.grouper.names, group_levels)
+                                                     names=labels.names)
 
-            for i, (name, labels) in enumerate(zipped):
-                result.insert(i, name, labels)
-            result = result.consolidate()
-        else:
-            index = self.grouper.result_index
-            result = DataFrame(output, index=index, columns=output_keys)
-
-        if self.axis == 1:
-            result = result.T
-
-        return result
+        return output_keys
 
     def _wrap_applied_output(self, keys, values, not_indexed_same=False):
         if len(keys) == 0:
@@ -1640,13 +1613,51 @@ def __getitem__(self, key):
                                     exclusions=self.exclusions,
                                     as_index=self.as_index)
 
+    def _wrap_generic_output(self, result, obj):
+        result_index = self.grouper.levels[0]
+
+        if result:
+            if self.axis == 0:
+                result = DataFrame(result, index=obj.columns,
+                                   columns=result_index).T
+            else:
+                result = DataFrame(result, index=obj.index,
+                                   columns=result_index)
+        else:
+            result = DataFrame(result)
+
+        return result
+
     def _get_data_to_aggregate(self):
         obj = self._obj_with_exclusions
         if self.axis == 1:
             return obj.T._data, 1
         else:
             return obj._data, 1
 
+    def _wrap_aggregated_output(self, output, names=None):
+        agg_axis = 0 if self.axis == 1 else 1
+        agg_labels = self._obj_with_exclusions._get_axis(agg_axis)
+
+        output_keys = self._decide_output_index(output, agg_labels)
+
+        if not self.as_index:
+            result = DataFrame(output, columns=output_keys)
+            group_levels = self.grouper.get_group_levels()
+            zipped = zip(self.grouper.names, group_levels)
+
+            for i, (name, labels) in enumerate(zipped):
+                result.insert(i, name, labels)
+            result = result.consolidate()
+        else:
+            index = self.grouper.result_index
+            result = DataFrame(output, index=index, columns=output_keys)
+
+        if self.axis == 1:
+            result = result.T
+
+        return result
+
     def _post_process_cython_aggregate(self, obj):
         # undoing kludge from below
         if self.axis == 0:
@@ -1733,31 +1744,58 @@ def aggregate(self, arg, *args, **kwargs):
 
         return self._aggregate_generic(arg, *args, **kwargs)
 
-    def _aggregate_generic(self, func, *args, **kwargs):
-        result = {}
+    def _wrap_generic_output(self, result, obj):
 
-        axis = self.axis
+        new_axes = list(obj.axes)
+        new_axes[self.axis] = self.grouper.result_index
+
+        result = Panel._from_axes(result, new_axes)
+
+        if self.axis > 0:
+            result = result.swapaxes(0, self.axis)
+
+        return result
 
+    def _aggregate_item_by_item(self, func, *args, **kwargs):
         obj = self._obj_with_exclusions
+        result = {}
+        cannot_agg = []
 
-        for name in self.grouper:
-            data = self.get_group(name, obj=obj)
-            try:
-                result[name] = func(data, *args, **kwargs)
-            except Exception:
-                wrapper = lambda x: func(x, *args, **kwargs)
-                result[name] = data.apply(wrapper, axis=axis)
+        if self.axis > 0:
+            for item in obj:
+                try:
+                    itemg = DataFrameGroupBy(obj[item],
+                                             axis=self.axis - 1,
+                                             grouper=self.grouper)
+                    result[item] = itemg.aggregate(func, *args, **kwargs)
+                except (ValueError, TypeError):
+                    raise
+                    # cannot_agg.append(item)
+                    # continue
+            new_axes = list(obj.axes)
+            new_axes[self.axis] = self.grouper.result_index
+            return Panel._from_axes(result, new_axes)
+        else:
+            raise NotImplementedError
 
-        result = Panel(result)
+    def _wrap_aggregated_output(self, output, names=None):
+        raise NotImplementedError
+        new_axes = list(self._obj_with_exclusions.axes)
+        new_axes[self.axis] = self.grouper.result_index
 
-        if axis > 0:
-            result = result.swapaxes(0, axis)
+        result = Panel(output, index=self.grouper.result_index,
+                       columns=output_keys)
+
+        if self.axis > 0:
+            result = result.swapaxes(0, self.axis)
 
         return result
 
+
 class NDArrayGroupBy(GroupBy):
     pass
 
+
 #----------------------------------------------------------------------
 # Grouping generator for BlockManager
 

pandas/tseries/resample.py

@@ -98,7 +98,7 @@ def _resample_timestamps(self, obj):
         # downsamples
         if len(grouper.binlabels) < len(axlabels):
             grouped  = obj.groupby(grouper, axis=self.axis)
-            result = grouped.agg(self.how)
+            result = grouped.aggregate(self.how)
         else:
             assert(self.axis == 0)
             # upsampling
@@ -139,7 +139,7 @@ def _resample_periods(self, obj):
             grouper = BinGrouper(bins, new_index)
 
             grouped = obj.groupby(grouper, axis=self.axis)
-            return grouped.agg(self.how)
+            return grouped.aggregate(self.how)
         elif is_superperiod(axlabels.freq, self.freq):
             # Get the fill indexer
             indexer = memb.get_indexer(new_index, method=self.fill_method,

pandas/tseries/tests/test_resample.py

@@ -347,6 +347,7 @@ def test_resample_panel_numpy(self):
         expected = panel.resample('M', how='mean', axis=1)
         tm.assert_panel_equal(result, expected)
 
+
 def _simple_ts(start, end, freq='D'):
     rng = date_range(start, end, freq=freq)
     return Series(np.random.randn(len(rng)), index=rng)